1 from pathlib
import Path
2 from typing
import Dict
, List
, Any
, Optional
3 from io
import StringIO
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
10 def _markdown_escape_char(char
: str) -> str:
15 if char
in "\\`*_{}[]()#+-.!":
20 def markdown_escape(v
: Any
) -> str:
21 return "".join([_markdown_escape_char(char
) for char
in str(v
)])
24 class DisplayStatus(enum
.Enum
):
26 NotYetStarted
= "Not yet started"
27 InProgress
= "Currently working on"
28 Completed
= "Completed but not yet added to payees list"
31 def from_status(status
: BugStatus
) -> "DisplayStatus":
32 return _DISPLAY_STATUS_MAP
[status
]
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
,
47 last_headers
: List
[str]
50 self
.buffer = StringIO()
51 self
.last_headers
= []
53 def write_headers(self
, headers
: List
[str]):
54 if headers
== self
.last_headers
:
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
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}",
79 if payment
is not None:
80 if node
.fixed_budget_excluding_subtasks \
81 != node
.budget_excluding_subtasks
:
82 total
= (f
"€{node.fixed_budget_excluding_subtasks} ("
83 f
"total is fixed from amount appearing in bug report,"
84 f
" which is €{node.budget_excluding_subtasks})")
86 total
= f
"€{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
" * €{payment.amount} out of total of {total}",
92 print(f
" * €{payment.amount} which is the total amount",
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
)
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
:
122 if len(node
.payments
) == 0 \
123 and node
.budget_excluding_subtasks
== 0 \
124 and node
.budget_including_subtasks
== 0:
127 headers
=[status_tracking_header
, display_status_header
],
128 node
=node
, payment
=None)
130 for display_status
in DisplayStatus
:
131 if display_status
== DisplayStatus
.Hidden \
132 or display_status
== DisplayStatus
.NotYetStarted
:
134 write_display_status_chunk(display_status
)
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"
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
,
151 node
=payment
.node
, payment
=payment
)
153 write_display_status_chunk(DisplayStatus
.NotYetStarted
)
155 return writer
.buffer.getvalue()
158 def write_budget_markdown(budget_graph
: BudgetGraph
,
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
,
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")