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}):\n  {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.submitted:
  88                 print(f"    * submitted on {payment.submitted}",
  89                       file=self.buffer)
  90             if payment.paid:
  91                 print(f"    * paid on {payment.paid}",
  92                       file=self.buffer)
  93             if payment.amount != node.fixed_budget_excluding_subtasks \
  94                     or payment.amount != node.budget_excluding_subtasks:
  95                 print(f"    * &euro;{payment.amount} out of total of {total}",
  96                       file=self.buffer)
  97             else:
  98                 print(f"    * &euro;{payment.amount} which is the total amount",
  99                       file=self.buffer)
 100
 101
 102 def _markdown_for_person(person: Person,
 103                          payments_dict: Dict[Milestone, List[Payment]],
 104                          assigned_nodes: List[Node]) -> str:
 105     writer = MarkdownWriter()
 106     print(f"<!-- autogenerated by budget-sync -->", file=writer.buffer)
 107     writer.write_headers([f"\n# {person.identifier}\n"])
 108     print(file=writer.buffer)
 109     status_tracking_header = "\n# Status Tracking\n"
 110     writer.write_headers([status_tracking_header])
 111     displayed_nodes_dict: Dict[DisplayStatus, List[Node]]
 112     displayed_nodes_dict = {i: [] for i in DisplayStatus}
 113     for node in assigned_nodes:
 114         display_status = DisplayStatus.from_status(node.status)
 115         displayed_nodes_dict[display_status].append(node)
 116
 117     def write_display_status_chunk(display_status: DisplayStatus):
 118         display_status_header = f"\n## {display_status.value}\n"
 119         for node in displayed_nodes_dict[display_status]:
 120             if display_status == DisplayStatus.Completed:
 121                 payment_found = False
 122                 for payment in node.payments.values():
 123                     if payment.payee == person:
 124                         payment_found = True
 125                         break
 126                 if payment_found:
 127                     continue
 128                 if len(node.payments) == 0 \
 129                         and node.budget_excluding_subtasks == 0 \
 130                         and node.budget_including_subtasks == 0:
 131                     continue
 132             writer.write_node(
 133                 headers=[status_tracking_header, display_status_header],
 134                 node=node, payment=None)
 135
 136     for display_status in DisplayStatus:
 137         if display_status == DisplayStatus.Hidden \
 138                 or display_status == DisplayStatus.NotYetStarted:
 139             continue
 140         write_display_status_chunk(display_status)
 141
 142     for payee_state in PayeeState:
 143         if payee_state == PayeeState.NotYetSubmitted:
 144             display_status_header = f"## Completed but not yet paid"
 145         elif payee_state == PayeeState.Submitted:
 146             display_status_header = f"## Submitted to NLNet but not yet paid"
 147         else:
 148             assert payee_state == PayeeState.Paid
 149             display_status_header = f"## Paid by NLNet"
 150         display_status_header = "\n%s\n" % display_status_header
 151         for milestone, payments_list in payments_dict.items():
 152             milestone_header = f"\n### {milestone.identifier}\n"
 153             for payment in payments_list:
 154                 if payment.state == payee_state:
 155                     writer.write_node(headers=[status_tracking_header,
 156                                                display_status_header,
 157                                                milestone_header],
 158                                       node=payment.node, payment=payment)
 159
 160     # write_display_status_chunk(DisplayStatus.NotYetStarted)
 161
 162     return writer.buffer.getvalue()
 163
 164
 165 def write_budget_markdown(budget_graph: BudgetGraph,
 166                           output_dir: Path):
 167     output_dir.mkdir(parents=True, exist_ok=True)
 168     for person, payments_dict in budget_graph.payments.items():
 169         markdown = _markdown_for_person(person,
 170                                         payments_dict,
 171                                         budget_graph.assigned_nodes[person])
 172         output_file = output_dir.joinpath(person.output_markdown_file)
 173         output_file.write_text(markdown, encoding="utf-8")