3 from datetime import datetime, timedelta
4 from urllib.parse import parse_qs, urlparse
5 from jinja2 import Environment as JinjaEnv, FileSystemLoader as JinjaFSLoader
6 from plomlib import PlomDB, PlomException, run_server, PlomHandler
7 from os.path import split as path_split
9 db_path = '/home/plom/org/ledger2024.dat'
11 j2env = JinjaEnv(loader=JinjaFSLoader('ledger_templates'))
13 class EditableException(PlomException):
14 def __init__(self, booking_index, *args, **kwargs):
15 self.booking_index = booking_index
16 super().__init__(*args, **kwargs)
21 def __init__(self, text_line):
22 self.text_line = text_line
24 split_by_comment = text_line.rstrip().split(sep=';', maxsplit=1)
25 self.non_comment = split_by_comment[0].rstrip()
26 self.empty = len(split_by_comment) == 1 and len(self.non_comment) == 0
29 if len(split_by_comment) == 2:
30 self.comment = split_by_comment[1].lstrip()
38 def __iadd__(self, moneys):
40 if type(moneys) == Wealth:
41 moneys = moneys.money_dict
42 for currency, amount in moneys.items():
43 if not currency in self.money_dict.keys():
44 self.money_dict[currency] = 0
45 self.money_dict[currency] += amount
50 return len(self.as_sink) == 0
55 for currency, amount in self.money_dict.items():
58 sink[currency] = -amount
64 def __init__(self, own_name, parent):
65 self.own_name = own_name
68 self.parent.children += [self]
69 self.own_moneys = Wealth()
72 def add_wealth(self, moneys):
73 self.own_moneys += moneys
77 if self.parent and len(self.parent.own_name) > 0:
78 return f'{self.parent.full_name}:{self.own_name}'
83 def full_moneys(self):
84 full_moneys = Wealth()
85 full_moneys += self.own_moneys
86 for child in self.children:
87 full_moneys += child.full_moneys
91 def parse_lines_to_bookings(lines, ignore_editable_exceptions=False):
92 lines = [LedgerTextLine(line) for line in lines]
93 lines += [LedgerTextLine('')] # to simulate ending of last booking
95 inside_booking = False
99 for i, line in enumerate(lines):
100 intro = f'file line {i}'
103 booking = Booking(lines=booking_lines, starts_at=booking_start_i)
104 if last_date > booking.date and not ignore_editable_exceptions:
105 raise EditableException(len(bookings), f'{intro}: out-of-order date (follows {last_date})')
106 last_date = booking.date
107 bookings += [booking]
108 inside_booking =False
111 if not inside_booking:
113 inside_booking = True
114 booking_lines += [line]
116 raise PlomException(f'{intro}: last booking unfinished')
122 def __init__(self, line=None, account='', amount=None, currency='', comment='', validate=True):
123 self.account = account
125 self.currency = currency
126 self.comment = comment
129 raise PlomException('line empty')
130 self.comment = line.comment
131 toks = line.non_comment.split()
132 if (len(toks) not in {1, 3}):
134 raise PlomException(f'number of non-comment tokens not 1 or 3')
139 self.account = toks[0]
142 self.amount = decimal.Decimal(toks[1])
143 except decimal.InvalidOperation:
145 raise PlomException(f'invalid token for Decimal: {toks[1]}')
147 self.comment = f'unparsed: {toks[1]}; {self.comment}'
148 self.currency = toks[2]
151 def amount_fmt(self):
152 if self.amount is None:
154 elif self.amount.as_tuple().exponent < -2:
155 return f'{self.amount:.1f}…'
157 return f'{self.amount:.2f}'
160 def for_writing(self):
161 line = f' {self.account}'
162 if self.amount is not None:
163 line += f' {self.amount} {self.currency}'
164 if self.comment != '':
165 line += f' ; {self.comment}'
169 def comment_cols(self):
170 return max(20, len(self.comment))
175 def __init__(self, lines=None, starts_at='?', date='', description='', top_comment='', transfer_lines=None, validate=True):
176 self.validate = validate
177 self.starts_at = starts_at
178 self.intro = f'booking starting at line {self.starts_at}'
185 self.description = description
186 self.top_comment = top_comment
188 self.transfer_lines = transfer_lines if transfer_lines else []
189 if self.validate and len(self.transfer_lines) < 2:
190 raise PlomException(f'{self.intro}: too few transfer lines')
191 self.calculate_account_changes()
192 self.lines = [LedgerTextLine(l) for l in self.for_writing]
195 def from_postvars(cls, postvars, starts_at='?', validate=True):
196 date = postvars['date'][0]
197 description = postvars['description'][0]
198 top_comment = postvars['top_comment'][0]
200 for i, account in enumerate(postvars['account']):
201 if len(account) == 0:
204 if len(postvars['amount'][i]) > 0:
205 amount = decimal.Decimal(postvars['amount'][i])
206 transfer_lines += [TransferLine(None, account, amount, postvars['currency'][i], postvars['comment'][i])]
207 return cls(None, starts_at, date, description, top_comment, transfer_lines, validate=validate)
210 self.transfer_lines = []
211 self.account_changes = {}
213 def parse_lines(self):
214 if len(self.lines) < 3 and self.validate:
215 raise PlomException(f'{self.intro}: ends with less than 3 lines:' + str(self.lines))
216 top_line = self.lines[0]
217 if top_line.empty and self.validate:
218 raise PlomException('{self.intro}: headline empty')
219 self.top_comment = top_line.comment
220 toks = top_line.non_comment.split(maxsplit=1)
223 raise PlomException(f'{self.intro}: headline missing elements: {non_comment}')
229 self.description = toks[1]
231 for i, line in enumerate(self.lines[1:]):
233 self.transfer_lines += [TransferLine(line, validate=self.validate)]
234 except PlomException as e:
235 raise PlomException(f'{self.intro}, transfer line {i}: {e}')
236 self.calculate_account_changes()
238 def calculate_account_changes(self):
240 money_changes = Wealth()
241 for i, transfer_line in enumerate(self.transfer_lines):
242 intro = f'{self.intro}, transfer line {i}'
243 if transfer_line.amount is None:
244 if sink_account is not None and self.validate:
245 raise PlomException(f'{intro}: second sink found (only one allowed)')
246 sink_account = transfer_line.account
248 if not transfer_line.account in self.account_changes.keys():
249 self.account_changes[transfer_line.account] = Wealth()
250 money = {transfer_line.currency: transfer_line.amount}
251 self.account_changes[transfer_line.account] += money
252 money_changes += money
253 if sink_account is None and (not money_changes.sink_empty) and self.validate:
254 raise PlomException(f'{intro}: does not balance (undeclared non-empty sink)')
255 if sink_account is not None:
256 if not sink_account in self.account_changes.keys():
257 self.account_changes[sink_account] = Wealth()
258 self.account_changes[sink_account] += money_changes.as_sink
261 def for_writing(self):
262 lines = [f'{self.date} {self.description}']
263 if self.top_comment is not None and self.top_comment.rstrip() != '':
264 lines[0] += f' ; {self.top_comment}'
265 for line in self.transfer_lines:
266 lines += [line.for_writing]
270 def comment_cols(self):
271 return max(20, len(self.top_comment))
273 def validate_head(self):
274 if not self.validate:
276 if len(self.date) == 0:
277 raise PlomException(f'{self.intro}: missing date')
278 if len(self.description) == 0:
279 raise PlomException(f'{self.intro}: missing description')
281 datetime.strptime(self.date, '%Y-%m-%d')
283 raise PlomException(f'{self.intro}: bad headline date format: {self.date}')
286 replacement_lines = []
287 for i, line in enumerate(self.transfer_lines):
288 if line.amount is None:
289 for currency, amount in self.account_changes[line.account].money_dict.items():
290 replacement_lines += [TransferLine(None, f'{line.account}', amount, currency).for_writing]
292 lines = self.lines[:i+1] + [LedgerTextLine(l) for l in replacement_lines] + self.lines[i+2:]
298 new_transfer_lines = []
299 for transfer_line in self.transfer_lines:
300 uncommented_source = LedgerTextLine(transfer_line.for_writing).non_comment
301 comment = f're: {uncommented_source.lstrip()}'
303 new_transfer_lines += [TransferLine(None, new_account, -transfer_line.amount, transfer_line.currency, comment)]
304 for transfer_line in new_transfer_lines:
305 self.lines += [LedgerTextLine(transfer_line.for_writing)]
309 def replace(self, replace_from, replace_to):
311 for l in self.for_writing:
312 lines += [l.replace(replace_from, replace_to)]
313 self.lines = [LedgerTextLine(l) for l in lines]
317 def add_taxes(self, db):
318 acc_kk_add = 'Reserves:KrankenkassenBeitragsWachstum'
319 acc_kk_minimum = 'Reserves:Monthly:KrankenkassenDefaultBeitrag'
320 acc_kk = 'Expenses:KrankenKasse'
321 acc_ESt = 'Reserves:EinkommensSteuer'
322 acc_assets = 'Assets'
323 acc_neuanfangspuffer_expenses = 'Reserves:NeuAnfangsPuffer:Ausgaben'
324 months_passed = datetime.strptime(self.date, '%Y-%m-%d').month - 1
327 past_neuanfangspuffer_expenses = 0
328 for b in db.bookings:
329 if b.date == self.date:
331 if acc_neuanfangspuffer_expenses in b.account_changes.keys():
332 past_neuanfangspuffer_expenses -= b.account_changes[acc_neuanfangspuffer_expenses].money_dict['€']
333 if acc_kk_add in b.account_changes.keys():
334 past_kk_add += b.account_changes[acc_kk_add].money_dict['€']
335 if acc_kk_minimum in b.account_changes.keys():
336 past_kk_expenses += b.account_changes[acc_kk_minimum].money_dict['€']
338 needed_netto = -self.account_changes['Assets'].money_dict['€']
339 past_taxed_needs_before_kk = past_neuanfangspuffer_expenses - past_kk_expenses
341 E0 = decimal.Decimal(11604)
342 E1 = decimal.Decimal(17006)
343 E2 = decimal.Decimal(66761)
344 E3 = decimal.Decimal(277826)
345 taxed_income_before_kk = needed_netto
347 too_high = 2 * needed_netto
349 estimate_for_remaining_year = (12 - months_passed) * taxed_income_before_kk
350 zvE = past_taxed_needs_before_kk + estimate_for_remaining_year
352 ESt_year = decimal.Decimal(0)
355 ESt_year = (decimal.Decimal(922.98) * y + 1400) * y
358 ESt_year = (decimal.Decimal(181.19) * y + 2397) * y + decimal.Decimal(1025.38)
360 ESt_year = decimal.Decimal(0.42) * zvE - 10602.13
362 ESt_year = decimal.Decimal(0.45) * zvE - 18936.88
363 ESt_this_month = ESt_year / 12
364 taxed_income_minus_ESt = taxed_income_before_kk - ESt_this_month
365 if abs(taxed_income_minus_ESt - needed_netto) < 0.001:
367 elif taxed_income_minus_ESt < needed_netto:
368 too_low = taxed_income_before_kk
369 elif taxed_income_minus_ESt > needed_netto:
370 too_high = taxed_income_before_kk
371 taxed_income_before_kk = too_low + (too_high - too_low)/2
372 ESt_this_month = ESt_this_month.quantize(decimal.Decimal('0.00'))
373 comment = f'estimated zvE: {past_taxed_needs_before_kk}€ + {estimate_for_remaining_year:.2f}€ = {zvE:.2f}€ → year ESt: {ESt_year:.2f} → needed taxed income before Krankenkasse: {taxed_income_before_kk:.2f}€'
374 self.transfer_lines += [TransferLine(None, acc_ESt, ESt_this_month, '€', comment)]
376 kk_factor = decimal.Decimal(1.197)
377 kk_minimum_income = 1178.33
378 kk_minimum_tax = decimal.Decimal(232.13).quantize(decimal.Decimal('0.00'))
379 if self.date < '2024-02-01':
380 kk_minimum_income = 1131.67
381 kk_minimum_tax = decimal.Decimal(222.94).quantize(decimal.Decimal('0.00'))
382 comment = f'assumed minimum income of {kk_minimum_income:.2f}€ * {kk_factor:.3f}'
383 self.transfer_lines += [TransferLine(None, acc_kk_minimum, kk_minimum_tax, '€', comment)]
384 kk_add = taxed_income_before_kk * kk_factor - taxed_income_before_kk - kk_minimum_tax
385 kk_add = decimal.Decimal(kk_add).quantize(decimal.Decimal('0.00'))
386 if past_kk_add + kk_add < 0: # *if* kk_add would actually kill all earlier kk_add …
387 kk_add = - past_kk_add # … shrink it so it won't push the kk_add total below 0
388 comment = f'local negative as large as possible without moving {acc_kk_add} below zero'
390 comment = f'({taxed_income_before_kk:.2f}€ * {kk_factor:.3f}) - {taxed_income_before_kk:.2f} - {kk_minimum_tax}'
391 self.transfer_lines += [TransferLine(None, acc_kk_add, kk_add, '€', comment)]
393 diff_through_taxes_and_kk = ESt_this_month + kk_minimum_tax + kk_add
394 comment = f'{ESt_this_month} + {kk_minimum_tax} + {kk_add}'
395 self.transfer_lines += [TransferLine(None, acc_assets, -diff_through_taxes_and_kk, '€', comment)]
396 final_loss = diff_through_taxes_and_kk + needed_netto
397 comment = f'{needed_netto} + {diff_through_taxes_and_kk}'
398 self.transfer_lines += [TransferLine(None, acc_assets, final_loss, '€', comment)]
399 self.transfer_lines += [TransferLine(None, acc_neuanfangspuffer_expenses, -final_loss, '€')]
402 class LedgerDB(PlomDB):
404 def __init__(self, prefix, ignore_editable_exceptions=False):
409 super().__init__(db_path)
410 self.bookings = parse_lines_to_bookings(self.text_lines, ignore_editable_exceptions)
412 def read_db_file(self, f):
413 self.text_lines += f.readlines()
415 def insert_booking_at_date(self, booking):
417 if len(self.bookings) > 0:
418 for i, iterated_booking in enumerate(self.bookings):
419 if booking.date < iterated_booking.date:
421 elif booking.date == iterated_booking.date:
426 self.bookings.insert(place_at, booking)
428 def ledger_as_html(self):
429 for index, booking in enumerate(self.bookings):
430 booking.can_up = index > 0 and self.bookings[index - 1].date == booking.date
431 booking.can_down = index < len(self.bookings) - 1 and self.bookings[index + 1].date == booking.date
432 return j2env.get_template('ledger.html').render(bookings=self.bookings)
434 def balance_as_html(self, until_after=None):
435 bookings = self.bookings[:(until_after if until_after is None else int(until_after)+1)]
436 account_trunk = Account('', None)
437 accounts = {account_trunk.full_name: account_trunk}
438 for booking in bookings:
439 for full_account_name, moneys in booking.account_changes.items():
440 toks = full_account_name.split(':')
443 parent_name = ':'.join(path)
445 account_name = ':'.join(path)
446 if not account_name in accounts.keys():
447 accounts[account_name] = Account(own_name=tok, parent=accounts[parent_name])
448 accounts[full_account_name].add_wealth(moneys)
450 def __init__(self, indent, name, moneys):
453 self.moneys = moneys.money_dict
455 def walk_tree(nodes, indent, account):
456 nodes += [Node(indent, account.own_name, account.full_moneys)]
457 for child in account.children:
458 walk_tree(nodes, indent+1, child)
459 for acc in account_trunk.children:
460 walk_tree(nodes, 0, acc)
461 return j2env.get_template('balance.html').render(nodes=nodes)
463 def edit(self, index, sent=None, error_msg=None, edit_mode='table', copy=False):
465 if sent or -1 == index:
466 content = sent if sent else ([] if 'textarea'==edit_mode else None)
468 content = self.bookings[index]
469 date_today = str(datetime.now())[:10]
471 content.date = date_today
473 elif -1 == index and (content is None or [] == content):
474 content = Booking(date=date_today, validate=False)
475 if 'textarea' == edit_mode and content:
476 content = content.for_writing
478 for booking in self.bookings:
479 for transfer_line in booking.transfer_lines:
480 accounts.add(transfer_line.account)
481 return j2env.get_template('edit.html').render(
489 def move_up(self, index):
490 return self.move(index, -1)
492 def move_down(self, index):
493 return self.move(index, +1)
495 def move(self, index, direction):
496 to_move = self.bookings[index]
497 swap_index = index + 1*(direction)
498 to_swap = self.bookings[swap_index]
499 self.bookings[index] = to_swap
500 self.bookings[index + 1*(direction)] = to_move
505 for i, booking in enumerate(self.bookings):
508 lines += booking.for_writing
509 self.write_text_to_db('\n'.join(lines) + '\n')
512 class LedgerHandler(PlomHandler):
514 def app_init(self, handler):
515 default_path = '/ledger'
516 handler.add_route('GET', default_path, self.forward_gets)
517 handler.add_route('POST', default_path, self.forward_posts)
518 return 'ledger', default_path
521 self.try_do(self.forward_posts)
523 def forward_posts(self):
524 prefix = self.apps['ledger'] if hasattr(self, 'apps') else ''
525 length = int(self.headers['content-length'])
526 postvars = parse_qs(self.rfile.read(length).decode(), keep_blank_values=1)
527 db = LedgerDB(prefix, ignore_editable_exceptions=True)
529 parsed_url = urlparse(self.path)
530 site = path_split(parsed_url.path)[1]
531 for string in {'update', 'add', 'check', 'mirror', 'fill_sink', 'textarea', 'table', 'move_up', 'move_down', 'add_taxes', 'replace'}:
532 if string in postvars.keys():
533 submit_button = string
535 if 'ledger' == site and submit_button in {'move_up', 'move_down'}:
536 mover = getattr(db, submit_button)
537 index = mover(int(postvars[submit_button][0]))
539 index = int(postvars['index'][0])
540 edit_mode = postvars['edit_mode'][0]
541 validate = submit_button in {'update', 'add', 'copy', 'check'}
542 starts_at = '?' if index == -1 else db.bookings[index].starts_at
544 if 'textarea' == edit_mode:
545 lines = [LedgerTextLine(line) for line in postvars['booking'][0].rstrip().split('\n')]
546 if len(lines) == 1 and lines[0].text_line == 'delete':
549 booking = Booking(lines, starts_at, validate=validate)
551 booking = Booking.from_postvars(postvars, starts_at, validate)
552 if submit_button in {'update', 'add'}:
553 if submit_button == 'update':
554 if 'textarea' == edit_mode and delete:
555 del db.bookings[index]
556 # if not creating new Booking, and date unchanged, keep it in place
557 elif booking.date == db.bookings[index].date:
558 db.bookings[index] = booking
560 del db.bookings[index]
561 db.insert_booking_at_date(booking)
563 db.insert_booking_at_date(booking)
564 else: # non-DB-writing calls
566 if 'check' == submit_button:
567 error_msg = 'All looks fine!'
568 elif submit_button in {'mirror', 'fill_sink', 'add_taxes'}:
569 if 'add_taxes' == submit_button:
570 booking.add_taxes(db)
572 getattr(booking, submit_button)()
573 elif 'replace' == submit_button:
574 booking.replace(postvars['replace_from'][0], postvars['replace_to'][0])
575 elif submit_button in {'textarea', 'table'}:
576 edit_mode = submit_button
577 page = db.edit(index, booking, error_msg=error_msg, edit_mode=edit_mode)
581 index = index if index >= 0 else len(db.bookings) - 1
582 self.redirect(f'ledger#{index}')
585 self.try_do(self.forward_gets)
587 def forward_gets(self):
588 prefix = self.apps['ledger'] if hasattr(self, 'apps') else ''
590 db = LedgerDB(prefix=prefix)
591 except EditableException as e:
592 # We catch the EditableException for further editing, and then
593 # re-run the DB initiation without it blocking DB creation.
594 db = LedgerDB(prefix=prefix, ignore_editable_exceptions=True)
595 page = db.edit(index=e.booking_index, error_msg=f'ERROR: {e}')
598 parsed_url = urlparse(self.path)
599 site = path_split(parsed_url.path)[1]
600 params = parse_qs(parsed_url.query)
601 if site == 'balance':
602 stop = params.get('until_after', [None])[0]
603 page = db.balance_as_html(stop)
605 index = params.get('i', [-1])[0]
606 copy = params.get('copy', [0])[0]
607 page = db.edit(int(index), copy=bool(copy))
609 page = db.ledger_as_html()
614 if __name__ == "__main__":
615 run_server(server_port, LedgerHandler)