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