1 from collections
import defaultdict
2 from pathlib
import Path
3 from typing
import Dict
, List
, Any
, Optional
4 from io
import StringIO
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
13 def _markdown_escape_char(char
: str) -> str:
18 if char
in "\\`*_{}[]()#+-.!":
23 def markdown_escape(v
: Any
) -> str:
24 return "".join([_markdown_escape_char(char
) for char
in str(v
)])
27 class DisplayStatus(enum
.Enum
):
29 NotYetStarted
= "Not yet started"
30 InProgress
= "Currently working on"
31 Completed
= "Completed but not yet added to payees list"
34 def from_status(status
: BugStatus
) -> "DisplayStatus":
35 return _DISPLAY_STATUS_MAP
[status
]
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
,
50 last_headers
: List
[str]
53 self
.buffer = StringIO()
54 self
.last_headers
= []
56 def write_headers(self
, headers
: List
[str]):
57 if headers
== self
.last_headers
:
59 for i
in range(len(headers
)):
60 if not headers
[i
].startswith("\n" + "#" * (i
+ 1) + " "):
62 "invalid markdown header. if you're not trying to make a"
63 " markdown header, don't use write_headers!")
64 if i
>= len(self
.last_headers
):
65 print(headers
[i
], file=self
.buffer)
66 self
.last_headers
.append(headers
[i
])
67 elif headers
[i
] != self
.last_headers
[i
]:
68 del self
.last_headers
[i
:]
69 print(headers
[i
], file=self
.buffer)
70 self
.last_headers
.append(headers
[i
])
71 if len(self
.last_headers
) > len(headers
):
72 raise ValueError("tried to go from deeper header scope stack to "
73 "ancestor scope without starting a new header, "
74 "which is not supported by markdown",
75 self
.last_headers
, headers
)
76 assert headers
== self
.last_headers
78 def write_node_header(self
,
80 node
: Optional
[Node
]):
81 self
.write_headers(headers
)
83 print("* None", file=self
.buffer)
85 summary
= markdown_escape(node
.bug
.summary
)
86 print(f
"* [Bug #{node.bug.id}]({node.bug_url}):\n {summary}",
92 payment
: Optional
[Payment
]):
93 self
.write_node_header(headers
, node
)
94 if payment
is not None:
95 if node
.fixed_budget_excluding_subtasks \
96 != node
.budget_excluding_subtasks
:
97 total
= (f
"€{node.fixed_budget_excluding_subtasks} ("
98 f
"total is fixed from amount appearing in bug report,"
99 f
" which is €{node.budget_excluding_subtasks})")
101 total
= f
"€{node.fixed_budget_excluding_subtasks}"
102 if payment
.submitted
:
103 print(f
" * submitted on {payment.submitted}",
106 print(f
" * paid on {payment.paid}",
108 if payment
.amount
!= node
.fixed_budget_excluding_subtasks \
109 or payment
.amount
!= node
.budget_excluding_subtasks
:
110 print(f
" * €{payment.amount} out of total of {total}",
113 print(f
" * €{payment.amount} which is the total amount",
115 closest
= node
.closest_bug_in_mou
117 print(f
" * this task is a MoU Milestone",
119 elif closest
is not None:
120 print(f
" * this task is part of MoU Milestone\n"
121 f
" [Bug #{closest.bug.id}]({closest.bug_url})",
123 elif payment
is not None: # only report this if there's a payment
124 print(f
" * neither this task nor any parent tasks are in "
129 def _markdown_for_person(person
: Person
,
130 payments_dict
: Dict
[Milestone
, List
[Payment
]],
131 assigned_nodes
: List
[Node
],
132 nodes_subset
: Optional
[OrderedSet
[Node
]] = None,
134 def node_included(node
: Node
) -> bool:
135 return nodes_subset
is None or node
in nodes_subset
136 writer
= MarkdownWriter()
137 print(f
"<!-- autogenerated by budget-sync -->", file=writer
.buffer)
138 writer
.write_headers([f
"\n# {person.full_name} ({person.identifier})\n"])
139 print(file=writer
.buffer)
140 status_tracking_header
= "\n# Status Tracking\n"
141 writer
.write_headers([status_tracking_header
])
142 displayed_nodes_dict
: Dict
[DisplayStatus
, List
[Node
]]
143 displayed_nodes_dict
= {i
: [] for i
in DisplayStatus
}
144 for node
in assigned_nodes
:
145 display_status
= DisplayStatus
.from_status(node
.status
)
146 displayed_nodes_dict
[display_status
].append(node
)
148 def write_display_status_chunk(display_status
: DisplayStatus
):
149 display_status_header
= f
"\n## {display_status.value}\n"
150 for node
in displayed_nodes_dict
[display_status
]:
151 if not node_included(node
):
153 if display_status
== DisplayStatus
.Completed
:
154 payment_found
= False
155 for payment
in node
.payments
.values():
156 if payment
.payee
== person
:
161 if len(node
.payments
) == 0 \
162 and node
.budget_excluding_subtasks
== 0 \
163 and node
.budget_including_subtasks
== 0:
166 headers
=[status_tracking_header
, display_status_header
],
167 node
=node
, payment
=None)
169 for display_status
in DisplayStatus
:
170 if display_status
== DisplayStatus
.Hidden \
171 or display_status
== DisplayStatus
.NotYetStarted
:
173 write_display_status_chunk(display_status
)
175 for payee_state
in PayeeState
:
176 # work out headers per status
177 if payee_state
== PayeeState
.NotYetSubmitted
:
178 display_status_header
= "\n## Payment not yet submitted\n"
179 subtotals_msg
= ("\nMoU Milestone subtotals for not "
180 "yet submitted payments\n")
181 elif payee_state
== PayeeState
.Submitted
:
182 display_status_header
= ("\n## Submitted to NLNet but "
184 subtotals_msg
= ("\nMoU Milestone subtotals for "
185 "submitted but not yet paid payments\n")
187 assert payee_state
== PayeeState
.Paid
188 display_status_header
= "\n## Paid by NLNet\n"
189 subtotals_msg
= ("\nMoU Milestone subtotals for paid "
191 # list all the payments grouped by Grant
192 for milestone
, payments_list
in payments_dict
.items():
193 milestone_header
= f
"\n### {milestone.identifier}\n"
194 mou_subtotals
: Dict
[Optional
[Node
], Money
] = defaultdict(Money
)
195 headers
= [status_tracking_header
,
196 display_status_header
,
198 # write out the payments and also compute the subtotals per
200 for payment
in payments_list
:
202 if payment
.state
== payee_state
and node_included(node
):
203 mou_subtotals
[node
.closest_bug_in_mou
] += payment
.amount
204 writer
.write_node(headers
=headers
,
205 node
=payment
.node
, payment
=payment
)
206 # now display the mou subtotals. really, this should be before
207 for node
, subtotal
in mou_subtotals
.items():
208 writer
.write_headers(headers
)
209 print(subtotals_msg
, file=writer
.buffer)
210 writer
.write_node_header(headers
, node
)
213 elif node
.fixed_budget_including_subtasks \
214 != node
.budget_including_subtasks
:
215 budget
= (" out of total including subtasks of "
216 f
"€{node.fixed_budget_including_subtasks}"
217 " (budget is fixed from amount appearing in "
218 "bug report, which is "
219 f
"€{node.budget_including_subtasks})")
221 budget
= (" out of total including subtasks of "
222 f
"€{node.fixed_budget_including_subtasks}")
223 print(f
" * subtotal €{subtotal}{budget}",
226 # write_display_status_chunk(DisplayStatus.NotYetStarted)
228 return writer
.buffer.getvalue()
231 def write_budget_markdown(budget_graph
: BudgetGraph
,
233 nodes_subset
: Optional
[OrderedSet
[Node
]] = None):
234 output_dir
.mkdir(parents
=True, exist_ok
=True)
235 for person
, payments_dict
in budget_graph
.payments
.items():
236 markdown
= _markdown_for_person(person
,
238 budget_graph
.assigned_nodes
[person
],
240 output_file
= output_dir
.joinpath(person
.output_markdown_file
)
241 output_file
.write_text(markdown
, encoding
="utf-8")
244 def markdown_for_person(budget_graph
: BudgetGraph
, person
: Person
,
245 nodes_subset
: Optional
[OrderedSet
[Node
]] = None,
247 return _markdown_for_person(person
, budget_graph
.payments
[person
],
248 budget_graph
.assigned_nodes
[person
],