from pathlib import Path
-from typing import Dict, List, Any
+from typing import Dict, List, Any, Optional
from io import StringIO
+import enum
from budget_sync.budget_graph import BudgetGraph, Node, Payment, PayeeState
from budget_sync.config import Person, Milestone, Config
+from budget_sync.util import BugStatus
def _markdown_escape_char(char: str) -> str:
return char
-def _markdown_escape(v: Any) -> str:
- return "".join([char for char in str(v)])
+def markdown_escape(v: Any) -> str:
+ return "".join([_markdown_escape_char(char) for char in str(v)])
+
+
+class DisplayStatus(enum.Enum):
+ Hidden = "Hidden"
+ NotYetStarted = "Not yet started"
+ InProgress = "Currently working on"
+ Completed = "Completed but not yet added to payees list"
+
+ @staticmethod
+ def from_status(status: BugStatus) -> "DisplayStatus":
+ return _DISPLAY_STATUS_MAP[status]
+
+
+_DISPLAY_STATUS_MAP = {
+ BugStatus.UNCONFIRMED: DisplayStatus.Hidden,
+ BugStatus.CONFIRMED: DisplayStatus.NotYetStarted,
+ BugStatus.IN_PROGRESS: DisplayStatus.InProgress,
+ BugStatus.DEFERRED: DisplayStatus.Hidden,
+ BugStatus.RESOLVED: DisplayStatus.Completed,
+ BugStatus.VERIFIED: DisplayStatus.Completed,
+ BugStatus.PAYMENTPENDING: DisplayStatus.Completed,
+}
+
+
+class MarkdownWriter:
+ last_headers: List[str]
+
+ def __init__(self):
+ self.buffer = StringIO()
+ self.last_headers = []
+
+ def write_headers(self, headers: List[str]):
+ if headers == self.last_headers:
+ return
+ for i in range(len(headers)):
+ if i >= len(self.last_headers):
+ print(headers[i], file=self.buffer)
+ self.last_headers.append(headers[i])
+ elif headers[i] != self.last_headers[i]:
+ del self.last_headers[i:]
+ print(headers[i], file=self.buffer)
+ self.last_headers.append(headers[i])
+ if len(self.last_headers) > len(headers):
+ raise ValueError("tried to go from deeper header scope stack to "
+ "ancestor scope without starting a new header, "
+ "which is not supported by markdown",
+ self.last_headers, headers)
+ assert headers == self.last_headers
+
+ def write_node(self,
+ headers: List[str],
+ node: Node,
+ payment: Optional[Payment]):
+ self.write_headers(headers)
+ summary = markdown_escape(node.bug.summary)
+ print(f"* [Bug #{node.bug.id}]({node.bug_url}): {summary}",
+ file=self.buffer)
+ if payment is not None:
+ if node.fixed_budget_excluding_subtasks \
+ != node.budget_excluding_subtasks:
+ total = (f"€{node.fixed_budget_excluding_subtasks} ("
+ f"total is fixed from amount appearing in bug report,"
+ f" which is €{node.budget_excluding_subtasks})")
+ else:
+ total = f"€{node.fixed_budget_excluding_subtasks}"
+ if payment.amount != node.fixed_budget_excluding_subtasks \
+ or payment.amount != node.budget_excluding_subtasks:
+ print(f" * €{payment.amount} out of total of {total}",
+ file=self.buffer)
+ else:
+ print(f" * €{payment.amount} which is the total amount",
+ file=self.buffer)
def _markdown_for_person(person: Person,
- payments_dict: Dict[Milestone, List[Payment]]) -> str:
- buffer = StringIO()
- print(f"<!-- autogenerated by budget-sync -->", file=buffer)
- print(f"# {person.identifier}", file=buffer)
- print(file=buffer)
- print(f"# Status Tracking", file=buffer)
- for milestone, payments_list in payments_dict.items():
- if len(payments_list) == 0:
+ payments_dict: Dict[Milestone, List[Payment]],
+ assigned_nodes: List[Node]) -> str:
+ writer = MarkdownWriter()
+ print(f"<!-- autogenerated by budget-sync -->", file=writer.buffer)
+ writer.write_headers([f"# {person.identifier}"])
+ print(file=writer.buffer)
+ status_tracking_header = "# Status Tracking"
+ writer.write_headers([status_tracking_header])
+ displayed_nodes_dict: Dict[DisplayStatus, List[Node]]
+ displayed_nodes_dict = {i: [] for i in DisplayStatus}
+ for node in assigned_nodes:
+ display_status = DisplayStatus.from_status(node.status)
+ displayed_nodes_dict[display_status].append(node)
+
+ def write_display_status_chunk(display_status: DisplayStatus):
+ display_status_header = f"## {display_status.value}"
+ for node in displayed_nodes_dict[display_status]:
+ if display_status == DisplayStatus.Completed:
+ payment_found = False
+ for payment in node.payments.values():
+ if payment.payee == person:
+ payment_found = True
+ break
+ if payment_found:
+ continue
+ if len(node.payments) == 0 \
+ and node.budget_excluding_subtasks == 0 \
+ and node.budget_including_subtasks == 0:
+ continue
+ writer.write_node(
+ headers=[status_tracking_header, display_status_header],
+ node=node, payment=None)
+
+ for display_status in DisplayStatus:
+ if display_status == DisplayStatus.Hidden \
+ or display_status == DisplayStatus.NotYetStarted:
continue
- print(f"## {milestone.identifier}", file=buffer)
- for payment in payments_list:
- # TODO: finish
- summary = _markdown_escape(payment.node.bug.summary)
- print(f"* [Bug #{payment.node.bug.id}]({payment.node.bug_url}): "
- f"{summary}",
- file=buffer)
- return buffer.getvalue()
+ write_display_status_chunk(display_status)
+
+ for payee_state in PayeeState:
+ if payee_state == PayeeState.NotYetSubmitted:
+ display_status_header = f"## Completed but not yet paid"
+ elif payee_state == PayeeState.Submitted:
+ display_status_header = f"## Submitted to NLNet but not yet paid"
+ else:
+ assert payee_state == PayeeState.Paid
+ display_status_header = f"## Paid by NLNet"
+ for milestone, payments_list in payments_dict.items():
+ milestone_header = f"### {milestone.identifier}"
+ for payment in payments_list:
+ if payment.state == payee_state:
+ writer.write_node(headers=[status_tracking_header,
+ display_status_header,
+ milestone_header],
+ node=payment.node, payment=payment)
+
+ write_display_status_chunk(DisplayStatus.NotYetStarted)
+
+ return writer.buffer.getvalue()
def write_budget_markdown(budget_graph: BudgetGraph,
output_dir: Path):
output_dir.mkdir(parents=True, exist_ok=True)
for person, payments_dict in budget_graph.payments.items():
- output_dir.joinpath(person.output_markdown_file) \
- .write_text(_markdown_for_person(person, payments_dict),
- encoding="utf-8")
+ markdown = _markdown_for_person(person,
+ payments_dict,
+ budget_graph.assigned_nodes[person])
+ output_file = output_dir.joinpath(person.output_markdown_file)
+ output_file.write_text(markdown, encoding="utf-8")