src/budget_sync/write_budget_markdown.py

   1 from pathlib import Path
   2 from typing import Dict, List, Any, Optional
   3 from io import StringIO
   4 import enum
   5 from budget_sync.budget_graph import BudgetGraph, Node, Payment, PayeeState
   6 from budget_sync.config import Person, Milestone, Config
   7 from budget_sync.util import BugStatus
   8
   9
  10 def _markdown_escape_char(char: str) -> str:
  11     if char == "<":
  12         return "&lt;"
  13     if char == "&":
  14         return "&amp;"
  15     if char in "\\`*_{}[]()#+-.!":
  16         return "\\" + char
  17     return char
  18
  19
  20 def markdown_escape(v: Any) -> str:
  21     return "".join([_markdown_escape_char(char) for char in str(v)])
  22
  23
  24 class DisplayStatus(enum.Enum):
  25     Hidden = "Hidden"
  26     NotYetStarted = "Not yet started"
  27     InProgress = "Currently working on"
  28     Completed = "Completed but not yet added to payees list"
  29
  30     @staticmethod
  31     def from_status(status: BugStatus) -> "DisplayStatus":
  32         return _DISPLAY_STATUS_MAP[status]
  33
  34
  35 _DISPLAY_STATUS_MAP = {
  36     BugStatus.UNCONFIRMED: DisplayStatus.Hidden,
  37     BugStatus.CONFIRMED: DisplayStatus.NotYetStarted,
  38     BugStatus.IN_PROGRESS: DisplayStatus.InProgress,
  39     BugStatus.DEFERRED: DisplayStatus.Hidden,
  40     BugStatus.RESOLVED: DisplayStatus.Completed,
  41     BugStatus.VERIFIED: DisplayStatus.Completed,
  42     BugStatus.PAYMENTPENDING: DisplayStatus.Completed,
  43 }
  44
  45
  46 class MarkdownWriter:
  47     last_headers: List[str]
  48
  49     def __init__(self):
  50         self.buffer = StringIO()
  51         self.last_headers = []
  52
  53     def write_headers(self, headers: List[str]):
  54         if headers == self.last_headers:
  55             return
  56         for i in range(len(headers)):
  57             if i >= len(self.last_headers):
  58                 print(headers[i], file=self.buffer)
  59                 self.last_headers.append(headers[i])
  60             elif headers[i] != self.last_headers[i]:
  61                 del self.last_headers[i:]
  62                 print(headers[i], file=self.buffer)
  63                 self.last_headers.append(headers[i])
  64         if len(self.last_headers) > len(headers):
  65             raise ValueError("tried to go from deeper header scope stack to "
  66                              "ancestor scope without starting a new header, "
  67                              "which is not supported by markdown",
  68                              self.last_headers, headers)
  69         assert headers == self.last_headers
  70
  71     def write_node(self,
  72                    headers: List[str],
  73                    node: Node,
  74                    payment: Optional[Payment]):
  75         self.write_headers(headers)
  76         summary = markdown_escape(node.bug.summary)
  77         print(f"* [Bug #{node.bug.id}]({node.bug_url}): {summary}",
  78               file=self.buffer)
  79         if payment is not None:
  80             if node.fixed_budget_excluding_subtasks \
  81                     != node.budget_excluding_subtasks:
  82                 total = (f"&euro;{node.fixed_budget_excluding_subtasks} ("
  83                          f"total is fixed from amount appearing in bug report,"
  84                          f" which is &euro;{node.budget_excluding_subtasks})")
  85             else:
  86                 total = f"&euro;{node.fixed_budget_excluding_subtasks}"
  87             if payment.amount != node.fixed_budget_excluding_subtasks \
  88                     or payment.amount != node.budget_excluding_subtasks:
  89                 print(f"    * &euro;{payment.amount} out of total of {total}",
  90                       file=self.buffer)
  91             else:
  92                 print(f"    * &euro;{payment.amount} which is the total amount",
  93                       file=self.buffer)
  94
  95
  96 def _markdown_for_person(person: Person,
  97                          payments_dict: Dict[Milestone, List[Payment]],
  98                          assigned_nodes: List[Node]) -> str:
  99     writer = MarkdownWriter()
 100     print(f"<!-- autogenerated by budget-sync -->", file=writer.buffer)
 101     writer.write_headers([f"# {person.identifier}"])
 102     print(file=writer.buffer)
 103     status_tracking_header = "# Status Tracking"
 104     writer.write_headers([status_tracking_header])
 105     displayed_nodes_dict: Dict[DisplayStatus, List[Node]]
 106     displayed_nodes_dict = {i: [] for i in DisplayStatus}
 107     for node in assigned_nodes:
 108         display_status = DisplayStatus.from_status(node.status)
 109         displayed_nodes_dict[display_status].append(node)
 110
 111     def write_display_status_chunk(display_status: DisplayStatus):
 112         display_status_header = f"## {display_status.value}"
 113         for node in displayed_nodes_dict[display_status]:
 114             if display_status == DisplayStatus.Completed:
 115                 payment_found = False
 116                 for payment in node.payments.values():
 117                     if payment.payee == person:
 118                         payment_found = True
 119                         break
 120                 if payment_found:
 121                     continue
 122                 if len(node.payments) == 0 \
 123                         and node.budget_excluding_subtasks == 0 \
 124                         and node.budget_including_subtasks == 0:
 125                     continue
 126             writer.write_node(
 127                 headers=[status_tracking_header, display_status_header],
 128                 node=node, payment=None)
 129
 130     for display_status in DisplayStatus:
 131         if display_status == DisplayStatus.Hidden \
 132                 or display_status == DisplayStatus.NotYetStarted:
 133             continue
 134         write_display_status_chunk(display_status)
 135
 136     for payee_state in PayeeState:
 137         if payee_state == PayeeState.NotYetSubmitted:
 138             display_status_header = f"## Completed but not yet paid"
 139         elif payee_state == PayeeState.Submitted:
 140             display_status_header = f"## Submitted to NLNet but not yet paid"
 141         else:
 142             assert payee_state == PayeeState.Paid
 143             display_status_header = f"## Paid by NLNet"
 144         for milestone, payments_list in payments_dict.items():
 145             milestone_header = f"### {milestone.identifier}"
 146             for payment in payments_list:
 147                 if payment.state == payee_state:
 148                     writer.write_node(headers=[status_tracking_header,
 149                                                display_status_header,
 150                                                milestone_header],
 151                                       node=payment.node, payment=payment)
 152
 153     # write_display_status_chunk(DisplayStatus.NotYetStarted)
 154
 155     return writer.buffer.getvalue()
 156
 157
 158 def write_budget_markdown(budget_graph: BudgetGraph,
 159                           output_dir: Path):
 160     output_dir.mkdir(parents=True, exist_ok=True)
 161     for person, payments_dict in budget_graph.payments.items():
 162         markdown = _markdown_for_person(person,
 163                                         payments_dict,
 164                                         budget_graph.assigned_nodes[person])
 165         output_file = output_dir.joinpath(person.output_markdown_file)
 166         output_file.write_text(markdown, encoding="utf-8")