src/budget_sync/write_budget_markdown.py

   1 from collections import defaultdict
   2 from pathlib import Path
   3 from typing import Dict, List, Any, Optional
   4 from io import StringIO
   5 import enum
   6 from budget_sync.budget_graph import BudgetGraph, Node, Payment, PayeeState, PaymentSummary
   7 from budget_sync.config import Person, Milestone
   8 from budget_sync.money import Money
   9 from budget_sync.ordered_set import OrderedSet
  10 from budget_sync.util import BugStatus
  11
  12
  13 def _markdown_escape_char(char: str) -> str:
  14     if char == "<":
  15         return "&lt;"
  16     if char == "&":
  17         return "&amp;"
  18     if char in "\\`*_{}[]()#+-.!":
  19         return "\\" + char
  20     return char
  21
  22
  23 def markdown_escape(v: Any) -> str:
  24     return "".join([_markdown_escape_char(char) for char in str(v)])
  25
  26
  27 class DisplayStatus(enum.Enum):
  28     Hidden = "Hidden"
  29     NotYetStarted = "Not yet started"
  30     InProgress = "Currently working on"
  31     Completed = "Completed but not yet added to payees list"
  32
  33     @staticmethod
  34     def from_status(status: BugStatus) -> "DisplayStatus":
  35         return _DISPLAY_STATUS_MAP[status]
  36
  37
  38 _DISPLAY_STATUS_MAP = {
  39     BugStatus.UNCONFIRMED: DisplayStatus.Hidden,
  40     BugStatus.CONFIRMED: DisplayStatus.NotYetStarted,
  41     BugStatus.IN_PROGRESS: DisplayStatus.InProgress,
  42     BugStatus.DEFERRED: DisplayStatus.Hidden,
  43     BugStatus.RESOLVED: DisplayStatus.Completed,
  44     BugStatus.VERIFIED: DisplayStatus.Completed,
  45     BugStatus.PAYMENTPENDING: DisplayStatus.Completed,
  46 }
  47
  48
  49 class MarkdownWriter:
  50     last_headers: List[str]
  51
  52     def __init__(self):
  53         self.buffer = StringIO()
  54         self.last_headers = []
  55
  56     def write_headers(self, headers: List[str]):
  57         if headers == self.last_headers:
  58             return
  59         for i in range(len(headers)):
  60             if i >= len(self.last_headers):
  61                 print(headers[i], file=self.buffer)
  62                 self.last_headers.append(headers[i])
  63             elif headers[i] != self.last_headers[i]:
  64                 del self.last_headers[i:]
  65                 print(headers[i], file=self.buffer)
  66                 self.last_headers.append(headers[i])
  67         if len(self.last_headers) > len(headers):
  68             raise ValueError("tried to go from deeper header scope stack to "
  69                              "ancestor scope without starting a new header, "
  70                              "which is not supported by markdown",
  71                              self.last_headers, headers)
  72         assert headers == self.last_headers
  73
  74     def write_node_header(self,
  75                           headers: List[str],
  76                           node: Optional[Node]):
  77         self.write_headers(headers)
  78         if node is None:
  79             print("* None", file=self.buffer)
  80             return
  81         summary = markdown_escape(node.bug.summary)
  82         print(f"* [Bug #{node.bug.id}]({node.bug_url}):\n  {summary}",
  83               file=self.buffer)
  84
  85     def write_node(self,
  86                    headers: List[str],
  87                    node: Node,
  88                    payment: Optional[Payment]):
  89         self.write_node_header(headers, node)
  90         if payment is not None:
  91             if node.fixed_budget_excluding_subtasks \
  92                     != node.budget_excluding_subtasks:
  93                 total = (f"&euro;{node.fixed_budget_excluding_subtasks} ("
  94                          f"total is fixed from amount appearing in bug report,"
  95                          f" which is &euro;{node.budget_excluding_subtasks})")
  96             else:
  97                 total = f"&euro;{node.fixed_budget_excluding_subtasks}"
  98             if payment.submitted:
  99                 print(f"    * submitted on {payment.submitted}",
 100                       file=self.buffer)
 101             if payment.paid:
 102                 print(f"    * paid on {payment.paid}",
 103                       file=self.buffer)
 104             if payment.amount != node.fixed_budget_excluding_subtasks \
 105                     or payment.amount != node.budget_excluding_subtasks:
 106                 print(f"    * &euro;{payment.amount} out of total of {total}",
 107                       file=self.buffer)
 108             else:
 109                 print(f"    * &euro;{payment.amount} which is the total amount",
 110                       file=self.buffer)
 111         closest = node.closest_bug_in_mou
 112         if closest is node:
 113             print(f"    * this task is a MoU Milestone",
 114                   file=self.buffer)
 115         elif closest is not None:
 116             print(f"    * this task is part of MoU Milestone\n"
 117                   f"      [Bug #{closest.bug.id}]({closest.bug_url})",
 118                   file=self.buffer)
 119         elif payment is not None:  # only report this if there's a payment
 120             print(f"    * neither this task nor any parent tasks are in "
 121                   f"the MoU",
 122                   file=self.buffer)
 123
 124
 125 def _markdown_for_person(person: Person,
 126                          payments_dict: Dict[Milestone, List[Payment]],
 127                          assigned_nodes: List[Node],
 128                          nodes_subset: Optional[OrderedSet[Node]] = None,
 129                          ) -> str:
 130     def node_included(node: Node) -> bool:
 131         return nodes_subset is None or node in nodes_subset
 132     writer = MarkdownWriter()
 133     print(f"<!-- autogenerated by budget-sync -->", file=writer.buffer)
 134     writer.write_headers([f"\n# {person.full_name} ({person.identifier})\n"])
 135     print(file=writer.buffer)
 136     status_tracking_header = "\n# Status Tracking\n"
 137     writer.write_headers([status_tracking_header])
 138     displayed_nodes_dict: Dict[DisplayStatus, List[Node]]
 139     displayed_nodes_dict = {i: [] for i in DisplayStatus}
 140     for node in assigned_nodes:
 141         display_status = DisplayStatus.from_status(node.status)
 142         displayed_nodes_dict[display_status].append(node)
 143
 144     def write_display_status_chunk(display_status: DisplayStatus):
 145         display_status_header = f"\n## {display_status.value}\n"
 146         for node in displayed_nodes_dict[display_status]:
 147             if not node_included(node):
 148                 continue
 149             if display_status == DisplayStatus.Completed:
 150                 payment_found = False
 151                 for payment in node.payments.values():
 152                     if payment.payee == person:
 153                         payment_found = True
 154                         break
 155                 if payment_found:
 156                     continue
 157                 if len(node.payments) == 0 \
 158                         and node.budget_excluding_subtasks == 0 \
 159                         and node.budget_including_subtasks == 0:
 160                     continue
 161             writer.write_node(
 162                 headers=[status_tracking_header, display_status_header],
 163                 node=node, payment=None)
 164
 165     for display_status in DisplayStatus:
 166         if display_status == DisplayStatus.Hidden \
 167                 or display_status == DisplayStatus.NotYetStarted:
 168             continue
 169         write_display_status_chunk(display_status)
 170
 171     for payee_state in PayeeState:
 172         # work out headers per status
 173         if payee_state == PayeeState.NotYetSubmitted:
 174             display_status_header = "\n## Payment not yet submitted\n"
 175             subtotals_header = ("\nMoU Milestone subtotals for not "
 176                                 "yet submitted payments\n")
 177         elif payee_state == PayeeState.Submitted:
 178             display_status_header = ("\n## Submitted to NLNet but "
 179                                      "not yet paid\n")
 180             subtotals_header = ("\nMoU Milestone subtotals for "
 181                                 "submitted but not yet paid payments\n")
 182         else:
 183             assert payee_state == PayeeState.Paid
 184             display_status_header = "\n## Paid by NLNet\n"
 185             subtotals_header = ("\nMoU Milestone subtotals for paid "
 186                                 "payments\n")
 187         # list all the payments grouped by Grant
 188         for milestone, payments_list in payments_dict.items():
 189             milestone_header = f"\n### {milestone.identifier}\n"
 190             mou_subtotals: Dict[Optional[Node], Money] = defaultdict(Money)
 191             headers = [status_tracking_header,
 192                        display_status_header,
 193                        milestone_header]
 194             # write out the payments and also compute the subtotals per
 195             # mou milestone
 196             for payment in payments_list:
 197                 node = payment.node
 198                 if payment.state == payee_state and node_included(node):
 199                     mou_subtotals[node.closest_bug_in_mou] += payment.amount
 200                     writer.write_node(headers=headers,
 201                                       node=payment.node, payment=payment)
 202             headers.append(subtotals_header)
 203             # now display the mou subtotals. really, this should be before
 204             for node, subtotal in mou_subtotals.items():
 205                 writer.write_node_header(headers, node)
 206                 if node is None:
 207                     budget = ""
 208                 elif node.fixed_budget_including_subtasks \
 209                         != node.budget_including_subtasks:
 210                     budget = (" out of total including subtasks of "
 211                               f"&euro;{node.fixed_budget_including_subtasks}"
 212                               " (budget is fixed from amount appearing in "
 213                               "bug report, which is "
 214                               f"&euro;{node.budget_including_subtasks})")
 215                 else:
 216                     budget = (" out of total including subtasks of "
 217                               f"&euro;{node.fixed_budget_including_subtasks}")
 218                 print(f"    * subtotal &euro;{subtotal}{budget}",
 219                       file=writer.buffer)
 220
 221     # write_display_status_chunk(DisplayStatus.NotYetStarted)
 222
 223     return writer.buffer.getvalue()
 224
 225
 226 def write_budget_markdown(budget_graph: BudgetGraph,
 227                           output_dir: Path,
 228                           nodes_subset: Optional[OrderedSet[Node]] = None):
 229     output_dir.mkdir(parents=True, exist_ok=True)
 230     for person, payments_dict in budget_graph.payments.items():
 231         markdown = _markdown_for_person(person,
 232                                         payments_dict,
 233                                         budget_graph.assigned_nodes[person],
 234                                         nodes_subset)
 235         output_file = output_dir.joinpath(person.output_markdown_file)
 236         output_file.write_text(markdown, encoding="utf-8")
 237
 238
 239 def markdown_for_person(budget_graph: BudgetGraph, person: Person,
 240                         nodes_subset: Optional[OrderedSet[Node]] = None,
 241                         ) -> str:
 242     return _markdown_for_person(person, budget_graph.payments[person],
 243                                 budget_graph.assigned_nodes[person],
 244                                 nodes_subset)