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
8 db_path = '/home/plom/org/ledger2024.dat'
10 j2env = JinjaEnv(loader=JinjaFSLoader('ledger_templates'))
12 class EditableException(PlomException):
13 def __init__(self, booking_index, *args, **kwargs):
14 self.booking_index = booking_index
15 super().__init__(*args, **kwargs)
20 def __init__(self, text_line):
21 self.text_line = text_line
23 split_by_comment = text_line.rstrip().split(sep=';', maxsplit=1)
24 self.non_comment = split_by_comment[0].rstrip()
25 self.empty = len(split_by_comment) == 1 and len(self.non_comment) == 0
28 if len(split_by_comment) == 2:
29 self.comment = split_by_comment[1].lstrip()
37 def __iadd__(self, moneys):
39 if type(moneys) == Wealth:
40 moneys = moneys.money_dict
41 for currency, amount in moneys.items():
42 if not currency in self.money_dict.keys():
43 self.money_dict[currency] = 0
44 self.money_dict[currency] += amount
49 return len(self.as_sink) == 0
54 for currency, amount in self.money_dict.items():
57 sink[currency] = -amount
63 def __init__(self, own_name, parent):
64 self.own_name = own_name
67 self.parent.children += [self]
68 self.own_moneys = Wealth()
71 def add_wealth(self, moneys):
72 self.own_moneys += moneys
76 if self.parent and len(self.parent.own_name) > 0:
77 return f'{self.parent.full_name}:{self.own_name}'
82 def full_moneys(self):
83 full_moneys = Wealth()
84 full_moneys += self.own_moneys
85 for child in self.children:
86 full_moneys += child.full_moneys
90 def parse_lines_to_bookings(lines, ignore_editable_exceptions=False):
91 lines = [LedgerTextLine(line) for line in lines]
92 lines += [LedgerTextLine('')] # to simulate ending of last booking
94 inside_booking = False
98 for i, line in enumerate(lines):
99 intro = f'file line {i}'
102 booking = Booking(lines=booking_lines, starts_at=booking_start_i)
103 if last_date > booking.date and not ignore_editable_exceptions:
104 raise EditableException(len(bookings), f'{intro}: out-of-order date (follows {last_date})')
105 last_date = booking.date
106 bookings += [booking]
107 inside_booking =False
110 if not inside_booking:
112 inside_booking = True
113 booking_lines += [line]
115 raise PlomException(f'{intro}: last booking unfinished')
121 def __init__(self, line=None, account='', amount=None, currency='', comment='', validate=True):
122 self.account = account
124 self.currency = currency
125 self.comment = comment
128 raise PlomException('line empty')
129 self.comment = line.comment
130 toks = line.non_comment.split()
131 if (len(toks) not in {1, 3}):
133 raise PlomException(f'number of non-comment tokens not 1 or 3')
138 self.account = toks[0]
141 self.amount = decimal.Decimal(toks[1])
142 except decimal.InvalidOperation:
144 raise PlomException(f'invalid token for Decimal: {toks[1]}')
146 self.comment = f'unparsed: {toks[1]}; {self.comment}'
147 self.currency = toks[2]
150 def amount_fmt(self):
151 if self.amount is None:
153 elif self.amount.as_tuple().exponent < -2:
154 return f'{self.amount:.1f}…'
156 return f'{self.amount:.2f}'
159 def for_writing(self):
160 line = f' {self.account}'
161 if self.amount is not None:
162 line += f' {self.amount} {self.currency}'
163 if self.comment != '':
164 line += f' ; {self.comment}'
168 def comment_cols(self):
169 return max(20, len(self.comment))
174 def __init__(self, lines=None, starts_at='?', date='', description='', top_comment='', transfer_lines=None, validate=True):
175 self.validate = validate
176 self.starts_at = starts_at
177 self.intro = f'booking starting at line {self.starts_at}'
184 self.description = description
185 self.top_comment = top_comment
187 self.transfer_lines = transfer_lines if transfer_lines else []
188 if self.validate and len(self.transfer_lines) < 2:
189 raise PlomException(f'{self.intro}: too few transfer lines')
190 self.calculate_account_changes()
191 self.lines = [LedgerTextLine(l) for l in self.for_writing]
194 def from_postvars(cls, postvars, starts_at='?', validate=True):
195 date = postvars['date'][0]
196 description = postvars['description'][0]
197 top_comment = postvars['top_comment'][0]
199 for i, account in enumerate(postvars['account']):
200 if len(account) == 0:
203 if len(postvars['amount'][i]) > 0:
204 amount = decimal.Decimal(postvars['amount'][i])
205 transfer_lines += [TransferLine(None, account, amount, postvars['currency'][i], postvars['comment'][i])]
206 return cls(None, starts_at, date, description, top_comment, transfer_lines, validate=validate)
209 self.transfer_lines = []
210 self.account_changes = {}
212 def parse_lines(self):
213 if len(self.lines) < 3 and self.validate:
214 raise PlomException(f'{self.intro}: ends with less than 3 lines:' + str(self.lines))
215 top_line = self.lines[0]
216 if top_line.empty and self.validate:
217 raise PlomException('{self.intro}: headline empty')
218 self.top_comment = top_line.comment
219 toks = top_line.non_comment.split(maxsplit=1)
222 raise PlomException(f'{self.intro}: headline missing elements: {non_comment}')
228 self.description = toks[1]
230 for i, line in enumerate(self.lines[1:]):
232 self.transfer_lines += [TransferLine(line, validate=self.validate)]
233 except PlomException as e:
234 raise PlomException(f'{self.intro}, transfer line {i}: {e}')
235 self.calculate_account_changes()
237 def calculate_account_changes(self):
239 money_changes = Wealth()
240 for i, transfer_line in enumerate(self.transfer_lines):
241 intro = f'{self.intro}, transfer line {i}'
242 if transfer_line.amount is None:
243 if sink_account is not None and self.validate:
244 raise PlomException(f'{intro}: second sink found (only one allowed)')
245 sink_account = transfer_line.account
247 if not transfer_line.account in self.account_changes.keys():
248 self.account_changes[transfer_line.account] = Wealth()
249 money = {transfer_line.currency: transfer_line.amount}
250 self.account_changes[transfer_line.account] += money
251 money_changes += money
252 if sink_account is None and (not money_changes.sink_empty) and self.validate:
253 raise PlomException(f'{intro}: does not balance (undeclared non-empty sink)')
254 if sink_account is not None:
255 if not sink_account in self.account_changes.keys():
256 self.account_changes[sink_account] = Wealth()
257 self.account_changes[sink_account] += money_changes.as_sink
260 def for_writing(self):
261 lines = [f'{self.date} {self.description}']
262 if self.top_comment is not None and self.top_comment.rstrip() != '':
263 lines[0] += f' ; {self.top_comment}'
264 for line in self.transfer_lines:
265 lines += [line.for_writing]
269 def comment_cols(self):
270 return max(20, len(self.top_comment))
272 def validate_head(self):
273 if not self.validate:
275 if len(self.date) == 0:
276 raise PlomException(f'{self.intro}: missing date')
277 if len(self.description) == 0:
278 raise PlomException(f'{self.intro}: missing description')
280 datetime.strptime(self.date, '%Y-%m-%d')
282 raise PlomException(f'{self.intro}: bad headline date format: {self.date}')
285 replacement_lines = []
286 for i, line in enumerate(self.transfer_lines):
287 if line.amount is None:
288 for currency, amount in self.account_changes[line.account].money_dict.items():
289 replacement_lines += [TransferLine(None, f'{line.account}', amount, currency).for_writing]
291 lines = self.lines[:i+1] + [LedgerTextLine(l) for l in replacement_lines] + self.lines[i+2:]
297 new_transfer_lines = []
298 for transfer_line in self.transfer_lines:
299 uncommented_source = LedgerTextLine(transfer_line.for_writing).non_comment
300 comment = f're: {uncommented_source.lstrip()}'
302 new_transfer_lines += [TransferLine(None, new_account, -transfer_line.amount, transfer_line.currency, comment)]
303 for transfer_line in new_transfer_lines:
304 self.lines += [LedgerTextLine(transfer_line.for_writing)]
308 def replace(self, replace_from, replace_to):
310 for l in self.for_writing:
311 lines += [l.replace(replace_from, replace_to)]
312 self.lines = [LedgerTextLine(l) for l in lines]
316 def add_taxes(self, db):
317 acc_kk_add = 'Reserves:KrankenkassenBeitragsWachstum'
318 acc_kk_minimum = 'Reserves:Monthly:KrankenkassenDefaultBeitrag'
319 acc_kk = 'Expenses:KrankenKasse'
320 acc_ESt = 'Reserves:EinkommensSteuer'
321 acc_assets = 'Assets'
322 acc_neuanfangspuffer_expenses = 'Reserves:NeuAnfangsPuffer:Ausgaben'
323 months_passed = datetime.strptime(self.date, '%Y-%m-%d').month - 1
326 past_neuanfangspuffer_expenses = 0
327 for b in db.bookings:
328 if b.date == self.date:
330 if acc_neuanfangspuffer_expenses in b.account_changes.keys():
331 past_neuanfangspuffer_expenses -= b.account_changes[acc_neuanfangspuffer_expenses].money_dict['€']
332 if acc_kk_add in b.account_changes.keys():
333 past_kk_add += b.account_changes[acc_kk_add].money_dict['€']
334 if acc_kk_minimum in b.account_changes.keys():
335 past_kk_expenses += b.account_changes[acc_kk_minimum].money_dict['€']
337 needed_netto = -self.account_changes['Assets'].money_dict['€']
338 past_taxed_needs_before_kk = past_neuanfangspuffer_expenses - past_kk_expenses
340 E0 = decimal.Decimal(11604)
341 E1 = decimal.Decimal(17006)
342 E2 = decimal.Decimal(66761)
343 E3 = decimal.Decimal(277826)
344 taxed_income_before_kk = needed_netto
346 too_high = 2 * needed_netto
348 estimate_for_remaining_year = (12 - months_passed) * taxed_income_before_kk
349 zvE = past_taxed_needs_before_kk + estimate_for_remaining_year
351 ESt_year = decimal.Decimal(0)
354 ESt_year = (decimal.Decimal(922.98) * y + 1400) * y
357 ESt_year = (decimal.Decimal(181.19) * y + 2397) * y + decimal.Decimal(1025.38)
359 ESt_year = decimal.Decimal(0.42) * zvE - 10602.13
361 ESt_year = decimal.Decimal(0.45) * zvE - 18936.88
362 ESt_this_month = ESt_year / 12
363 taxed_income_minus_ESt = taxed_income_before_kk - ESt_this_month
364 if abs(taxed_income_minus_ESt - needed_netto) < 0.001:
366 elif taxed_income_minus_ESt < needed_netto:
367 too_low = taxed_income_before_kk
368 elif taxed_income_minus_ESt > needed_netto:
369 too_high = taxed_income_before_kk
370 taxed_income_before_kk = too_low + (too_high - too_low)/2
371 ESt_this_month = ESt_this_month.quantize(decimal.Decimal('0.00'))
372 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}€'
373 self.transfer_lines += [TransferLine(None, acc_ESt, ESt_this_month, '€', comment)]
375 kk_factor = decimal.Decimal(1.197)
376 kk_minimum_income = 1178.33
377 kk_minimum_tax = decimal.Decimal(232.13).quantize(decimal.Decimal('0.00'))
378 if self.date < '2024-02-01':
379 kk_minimum_income = 1131.67
380 kk_minimum_tax = decimal.Decimal(222.94).quantize(decimal.Decimal('0.00'))
381 comment = f'assumed minimum income of {kk_minimum_income:.2f}€ * {kk_factor:.3f}'
382 self.transfer_lines += [TransferLine(None, acc_kk_minimum, kk_minimum_tax, '€', comment)]
383 kk_add = taxed_income_before_kk * kk_factor - taxed_income_before_kk - kk_minimum_tax
384 kk_add = decimal.Decimal(kk_add).quantize(decimal.Decimal('0.00'))
385 if past_kk_add + kk_add < 0: # *if* kk_add would actually kill all earlier kk_add …
386 kk_add = - past_kk_add # … shrink it so it won't push the kk_add total below 0
387 comment = f'local negative as large as possible without moving {acc_kk_add} below zero'
389 comment = f'({taxed_income_before_kk:.2f}€ * {kk_factor:.3f}) - {taxed_income_before_kk:.2f} - {kk_minimum_tax}'
390 self.transfer_lines += [TransferLine(None, acc_kk_add, kk_add, '€', comment)]
392 diff_through_taxes_and_kk = ESt_this_month + kk_minimum_tax + kk_add
393 comment = f'{ESt_this_month} + {kk_minimum_tax} + {kk_add}'
394 self.transfer_lines += [TransferLine(None, acc_assets, -diff_through_taxes_and_kk, '€', comment)]
395 final_loss = diff_through_taxes_and_kk + needed_netto
396 comment = f'{needed_netto} + {diff_through_taxes_and_kk}'
397 self.transfer_lines += [TransferLine(None, acc_assets, final_loss, '€', comment)]
398 self.transfer_lines += [TransferLine(None, acc_neuanfangspuffer_expenses, -final_loss, '€')]
401 class LedgerDB(PlomDB):
403 def __init__(self, prefix, ignore_editable_exceptions=False):
408 super().__init__(db_path)
409 self.bookings = parse_lines_to_bookings(self.text_lines, ignore_editable_exceptions)
411 def read_db_file(self, f):
412 self.text_lines += f.readlines()
414 def insert_booking_at_date(self, booking):
416 if len(self.bookings) > 0:
417 for i, iterated_booking in enumerate(self.bookings):
418 if booking.date < iterated_booking.date:
420 elif booking.date == iterated_booking.date:
425 self.bookings.insert(place_at, booking)
427 def ledger_as_html(self):
428 for index, booking in enumerate(self.bookings):
429 booking.can_up = index > 0 and self.bookings[index - 1].date == booking.date
430 booking.can_down = index < len(self.bookings) - 1 and self.bookings[index + 1].date == booking.date
431 return j2env.get_template('ledger.html').render(bookings=self.bookings)
433 def balance_as_html(self, until_after=None):
434 bookings = self.bookings[:(until_after if until_after is None else int(until_after)+1)]
435 account_trunk = Account('', None)
436 accounts = {account_trunk.full_name: account_trunk}
437 for booking in bookings:
438 for full_account_name, moneys in booking.account_changes.items():
439 toks = full_account_name.split(':')
442 parent_name = ':'.join(path)
444 account_name = ':'.join(path)
445 if not account_name in accounts.keys():
446 accounts[account_name] = Account(own_name=tok, parent=accounts[parent_name])
447 accounts[full_account_name].add_wealth(moneys)
449 def __init__(self, indent, name, moneys):
452 self.moneys = moneys.money_dict
454 def walk_tree(nodes, indent, account):
455 nodes += [Node(indent, account.own_name, account.full_moneys)]
456 for child in account.children:
457 walk_tree(nodes, indent+1, child)
458 for acc in account_trunk.children:
459 walk_tree(nodes, 0, acc)
460 return j2env.get_template('balance.html').render(nodes=nodes)
462 def edit(self, index, sent=None, error_msg=None, edit_mode='table', copy=False):
464 if sent or -1 == index:
465 content = sent if sent else ([] if 'textarea'==edit_mode else None)
467 content = self.bookings[index]
468 date_today = str(datetime.now())[:10]
470 content.date = date_today
471 elif -1 == index and (content is None or [] == content):
472 content = Booking(date=date_today, validate=False)
473 if 'textarea' == edit_mode and content:
474 content = content.for_writing
476 for booking in self.bookings:
477 for transfer_line in booking.transfer_lines:
478 accounts.add(transfer_line.account)
479 return j2env.get_template('edit.html').render(content=content, index=index, error_msg=error_msg, edit_mode=edit_mode, accounts=accounts, adding=(copy or -1 == index))
481 def move_up(self, index):
482 return self.move(index, -1)
484 def move_down(self, index):
485 return self.move(index, +1)
487 def move(self, index, direction):
488 to_move = self.bookings[index]
489 swap_index = index + 1*(direction)
490 to_swap = self.bookings[swap_index]
491 self.bookings[index] = to_swap
492 self.bookings[index + 1*(direction)] = to_move
497 for i, booking in enumerate(self.bookings):
500 lines += booking.for_writing
501 self.write_text_to_db('\n'.join(lines) + '\n')
504 class LedgerHandler(PlomHandler):
506 def app_init(self, handler):
507 default_path = '/ledger'
508 handler.add_route('GET', default_path, self.forward_gets)
509 handler.add_route('POST', default_path, self.forward_posts)
510 return 'ledger', default_path
513 self.try_do(self.forward_posts)
515 def forward_posts(self):
516 prefix = self.apps['ledger'] if hasattr(self, 'apps') else ''
517 length = int(self.headers['content-length'])
518 postvars = parse_qs(self.rfile.read(length).decode(), keep_blank_values=1)
519 db = LedgerDB(prefix, ignore_editable_exceptions=True)
521 parsed_url = urlparse(self.path)
522 for string in {'update', 'add', 'check', 'mirror', 'fill_sink', 'textarea', 'table', 'move_up', 'move_down', 'add_taxes', 'replace'}:
523 if string in postvars.keys():
524 submit_button = string
526 if f'{prefix}/ledger' == parsed_url.path and submit_button in {'move_up', 'move_down'}:
527 mover = getattr(db, submit_button)
528 index = mover(int(postvars[submit_button][0]))
529 elif prefix + '/edit' == parsed_url.path:
530 index = int(postvars['index'][0])
531 edit_mode = postvars['edit_mode'][0]
532 validate = submit_button in {'update', 'add', 'copy', 'check'}
533 starts_at = '?' if index == -1 else db.bookings[index].starts_at
535 if 'textarea' == edit_mode:
536 lines = [LedgerTextLine(line) for line in postvars['booking'][0].rstrip().split('\n')]
537 if len(lines) == 1 and lines[0].text_line == 'delete':
540 booking = Booking(lines, starts_at, validate=validate)
542 booking = Booking.from_postvars(postvars, starts_at, validate)
543 if submit_button in {'update', 'add'}:
544 if submit_button == 'update':
545 # if 'textarea' == edit_mode and 'delete' == ''.join([l.text_line for l in lines]).strip():
546 if 'textarea' == edit_mode and delete:
547 del db.bookings[index]
548 # if not creating new Booking, and date unchanged, keep it in place
549 elif booking.date == db.bookings[index].date:
550 db.bookings[index] = booking
552 del db.bookings[index]
553 db.insert_booking_at_date(booking)
555 db.insert_booking_at_date(booking)
556 else: # non-DB-writing calls
558 if 'check' == submit_button:
559 error_msg = 'All looks fine!'
560 elif submit_button in {'mirror', 'fill_sink', 'add_taxes'}:
561 if 'add_taxes' == submit_button:
562 booking.add_taxes(db)
564 getattr(booking, submit_button)()
565 elif 'replace' == submit_button:
566 booking.replace(postvars['replace_from'][0], postvars['replace_to'][0])
567 elif submit_button in {'textarea', 'table'}:
568 edit_mode = submit_button
569 page = db.edit(index, booking, error_msg=error_msg, edit_mode=edit_mode)
573 index = index if index >= 0 else len(db.bookings) - 1
574 self.redirect(prefix + f'/ledger#{index}')
577 self.try_do(self.forward_gets)
579 def forward_gets(self):
580 prefix = self.apps['ledger'] if hasattr(self, 'apps') else ''
582 db = LedgerDB(prefix=prefix)
583 except EditableException as e:
584 # We catch the EditableException for further editing, and then
585 # re-run the DB initiation without it blocking DB creation.
586 db = LedgerDB(prefix=prefix, ignore_editable_exceptions=True)
587 page = db.edit(index=e.booking_index, error_msg=f'ERROR: {e}')
590 parsed_url = urlparse(self.path)
591 params = parse_qs(parsed_url.query)
592 if parsed_url.path == f'{prefix}/balance':
593 stop = params.get('until_after', [None])[0]
594 page = db.balance_as_html(stop)
595 elif parsed_url.path == f'{prefix}/edit':
596 index = params.get('i', [-1])[0]
597 copy = params.get('copy', [0])[0]
598 page = db.edit(int(index), copy=bool(copy))
600 page = db.ledger_as_html()
605 if __name__ == "__main__":
606 run_server(server_port, LedgerHandler)