adjust/rewrite code to fix https://bugs.libre-soc.org/show_bug.cgi?id=706

[utils.git] / src / budget_sync / write_budget_csv.py

diff --git a/src/budget_sync/write_budget_csv.py b/src/budget_sync/write_budget_csv.py
new file mode 100644 (file)
index 0000000..c040421
--- /dev/null
+++ b/src/budget_sync/write_budget_csv.py
@@ -0,0 +1,84 @@
+import csv
+from enum import Enum, auto
+from io import StringIO
+from typing import Any, Callable, Dict, List, Optional
+from budget_sync.budget_graph import BudgetGraph, Node, PayeeState, PaymentSummary
+from pathlib import Path
+from budget_sync.config import Milestone
+from budget_sync.money import Money
+from budget_sync.write_budget_markdown import markdown_escape
+
+
+def _budget_csv_row(budget_graph: BudgetGraph, milestone: Milestone, node: Optional[Node]) -> Dict[str, str]:
+    row_fns: Dict[str, Callable[[Node], Any]] = {
+        'bug_id': lambda node: node.bug.id,
+        'budget_excluding_subtasks': lambda node: node.budget_excluding_subtasks,
+        'budget_including_subtasks': lambda node: node.budget_including_subtasks,
+        'fixed_budget_excluding_subtasks': lambda node: node.fixed_budget_excluding_subtasks,
+        'fixed_budget_including_subtasks': lambda node: node.fixed_budget_including_subtasks,
+        'submitted_excluding_subtasks': lambda node: node.submitted_excluding_subtasks,
+        'paid_excluding_subtasks': lambda node: node.paid_excluding_subtasks,
+    }
+    milestone_people = budget_graph.milestone_people[milestone]
+
+    def handle_person(person):
+        # need a nested function in order to create a new person variable
+        # for this iteration that can be bound to the lambdas
+        id = person.identifier
+        row_fns.update({
+            id + " (planned amt)": lambda node: node.payment_summaries[person].total,
+            id + " (req amt)": lambda node: node.payment_summaries[person].total_submitted,
+            id + " (req date)": lambda node: node.payment_summaries[person].submitted_date,
+            id + " (paid amt)": lambda node: node.payment_summaries[person].total_paid,
+            id + " (paid date)": lambda node: node.payment_summaries[person].paid_date,
+        })
+    for person in milestone_people:
+        handle_person(person)
+    row = {k: "" for k in row_fns.keys()}
+    if node is None:
+        return row
+    for k, fn in row_fns.items():
+        try:
+            v = fn(node)
+        except KeyError:
+            continue
+        if v is not None:
+            row[k] = str(v)
+    return row
+
+
+def _budget_csv_for_milestone(budget_graph: BudgetGraph, milestone: Milestone) -> str:
+    with StringIO() as string_io:
+        writer = csv.DictWriter(
+            string_io,
+            _budget_csv_row(budget_graph, milestone, None).keys(),
+            lineterminator="\n")
+        writer.writeheader()
+        for node in budget_graph.assigned_nodes_for_milestones[milestone]:
+            # skip uninteresting nodes
+            if len(node.payments) == 0 \
+                    and node.budget_excluding_subtasks == 0 \
+                    and node.budget_including_subtasks == 0:
+                continue
+            row = _budget_csv_row(budget_graph, milestone, node)
+            writer.writerow(row)
+        return string_io.getvalue()
+
+
+def write_budget_csv(budget_graph: BudgetGraph,
+                     output_dir: Path):
+    output_dir.mkdir(parents=True, exist_ok=True)
+    milestones = budget_graph.config.milestones
+    csv_paths: Dict[Milestone, Path] = {}
+    for milestone in milestones.values():
+        csv_text = _budget_csv_for_milestone(budget_graph, milestone)
+        csv_paths[milestone] = output_dir.joinpath(
+            f"{milestone.identifier}.csv")
+        csv_paths[milestone].write_text(csv_text, encoding="utf-8")
+
+    markdown_text = "\n".join(f"# {markdown_escape(milestone.identifier)}\n"
+                              "\n"
+                              f"[[!table format=csv file=\"{path!s}\"]]"
+                              for milestone, path in csv_paths.items())
+    output_dir.joinpath("csvs.mdwn").write_text(
+        markdown_text, encoding="utf-8")