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}):\n {summary}",
79 closest
= node
.closest_bug_in_mou
81 print(f
" * this task is an MoU Milestone",
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})",
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 "
91 if payment
is not None:
92 if node
.fixed_budget_excluding_subtasks \
93 != node
.budget_excluding_subtasks
:
94 total
= (f
"€{node.fixed_budget_excluding_subtasks} ("
95 f
"total is fixed from amount appearing in bug report,"
96 f
" which is €{node.budget_excluding_subtasks})")
98 total
= f
"€{node.fixed_budget_excluding_subtasks}"
100 print(f
" * submitted on {payment.submitted}",
103 print(f
" * paid on {payment.paid}",
105 if payment
.amount
!= node
.fixed_budget_excluding_subtasks \
106 or payment
.amount
!= node
.budget_excluding_subtasks
:
107 print(f
" * €{payment.amount} out of total of {total}",
110 print(f
" * €{payment.amount} which is the total amount",
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
)
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
:
140 if len(node
.payments
) == 0 \
141 and node
.budget_excluding_subtasks
== 0 \
142 and node
.budget_including_subtasks
== 0:
145 headers
=[status_tracking_header
, display_status_header
],
146 node
=node
, payment
=None)
148 for display_status
in DisplayStatus
:
149 if display_status
== DisplayStatus
.Hidden \
150 or display_status
== DisplayStatus
.NotYetStarted
:
152 write_display_status_chunk(display_status
)
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"
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
,
170 node
=payment
.node
, payment
=payment
)
172 # write_display_status_chunk(DisplayStatus.NotYetStarted)
174 return writer
.buffer.getvalue()
177 def write_budget_markdown(budget_graph
: BudgetGraph
,
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
,
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")