home · contact · privacy
656a26e95656a6d83e6c23a7814be7f648728076
[misc] / ledger.py
1 import os
2 import decimal
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
8 db_path = '/home/plom/org/ledger2024.dat'
9 server_port = 8082
10 j2env = JinjaEnv(loader=JinjaFSLoader('ledger_templates'))
11
12 class EditableException(PlomException):
13     def __init__(self, booking_index, *args, **kwargs):
14         self.booking_index = booking_index
15         super().__init__(*args, **kwargs)
16     
17
18 class LedgerTextLine:
19
20     def __init__(self, text_line):
21         self.text_line = text_line
22         self.comment = '' 
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
26         if self.empty:
27             return
28         if len(split_by_comment) == 2:
29             self.comment = split_by_comment[1].lstrip()
30
31
32 class Wealth:
33
34     def __init__(self):
35         self.money_dict = {}
36
37     def __iadd__(self, moneys):
38         money_dict = 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
45         return self
46
47     @property
48     def sink_empty(self):
49         return len(self.as_sink) == 0
50
51     @property
52     def as_sink(self):
53         sink = {} 
54         for currency, amount in self.money_dict.items():
55             if 0 == amount:
56                 continue
57             sink[currency] = -amount
58         return sink 
59
60
61 class Account:
62
63     def __init__(self, own_name, parent):
64         self.own_name = own_name
65         self.parent = parent
66         if self.parent:
67             self.parent.children += [self]
68         self.own_moneys = Wealth() 
69         self.children = []
70
71     def add_wealth(self, moneys):
72         self.own_moneys += moneys
73
74     @property
75     def full_name(self):
76         if self.parent and len(self.parent.own_name) > 0:
77             return f'{self.parent.full_name}:{self.own_name}'
78         else:
79             return self.own_name
80
81     @property
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
87         return full_moneys
88
89
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
93     bookings = []
94     inside_booking = False
95     booking_lines = []
96     booking_start_i = 0
97     last_date = '' 
98     for i, line in enumerate(lines):
99         intro = f'file line {i}'
100         if line.empty: 
101             if inside_booking:
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
108                 booking_lines = []
109         else:
110             if not inside_booking:
111                 booking_start_i = i
112             inside_booking = True
113             booking_lines += [line]
114     if inside_booking:
115         raise PlomException(f'{intro}: last booking unfinished')
116     return bookings
117
118
119 class TransferLine:
120
121     def __init__(self, line=None, account='', amount=None, currency='', comment='', validate=True):
122         self.account = account 
123         self.amount = amount 
124         self.currency = currency 
125         self.comment = comment 
126         if line:
127             if line.empty:
128                 raise PlomException('line empty')
129             self.comment = line.comment
130             toks = line.non_comment.split()
131             if (len(toks) not in {1, 3}):
132                 if validate:
133                     raise PlomException(f'number of non-comment tokens not 1 or 3')
134                 elif len(toks) == 2:
135                     toks += ['']
136                 else:
137                     toks = 3*['']
138             self.account = toks[0]
139             if len(toks) != 1:
140                 try:
141                     self.amount = decimal.Decimal(toks[1])
142                 except decimal.InvalidOperation:
143                     if validate:
144                         raise PlomException(f'invalid token for Decimal: {toks[1]}')
145                     else:
146                         self.comment = f'unparsed: {toks[1]}; {self.comment}'
147                 self.currency = toks[2]
148
149     @property
150     def amount_fmt(self):
151         if self.amount is None:
152             return ''
153         elif self.amount.as_tuple().exponent < -2:
154             return f'{self.amount:.1f}…'
155         else:
156             return f'{self.amount:.2f}'
157
158     @property
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}' 
165         return line
166
167     @property
168     def comment_cols(self):
169         return max(20, len(self.comment))
170
171
172 class Booking:
173
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}'
178         self.clean()
179         if lines:
180             self.lines = lines 
181             self.parse_lines()
182         else:
183             self.date = date
184             self.description = description
185             self.top_comment = top_comment 
186             self.validate_head()
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]
192
193     @classmethod
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] 
198         transfer_lines = []
199         for i, account in enumerate(postvars['account']):
200             if len(account) == 0:
201                 continue
202             amount = None
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)
207
208     def clean(self):
209         self.transfer_lines = []
210         self.account_changes = {}
211
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)
220         if len(toks) < 2:
221             if self.validate:
222                 raise PlomException(f'{self.intro}: headline missing elements: {non_comment}')
223             elif 0 == len(toks):
224                 toks = 2*['']
225             else:
226                 toks += ['']
227         self.date = toks[0]
228         self.description = toks[1]
229         self.validate_head()
230         for i, line in enumerate(self.lines[1:]):
231             try:
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()
236
237     def calculate_account_changes(self):
238         sink_account = None 
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
246             else:
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
258
259     @property
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]
266         return lines
267
268     @property
269     def comment_cols(self):
270         return max(20, len(self.top_comment))
271
272     def validate_head(self):
273         if not self.validate:
274             return
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')
279         try:
280             datetime.strptime(self.date, '%Y-%m-%d')
281         except ValueError:
282             raise PlomException(f'{self.intro}: bad headline date format: {self.date}')
283
284     def fill_sink(self):
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]
290                 break
291         lines = self.lines[:i+1] + [LedgerTextLine(l) for l in replacement_lines] + self.lines[i+2:]
292         self.clean()
293         self.lines = lines
294         self.parse_lines()
295
296     def mirror(self):
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()}'
301             new_account = '?'
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)]
305         self.clean()
306         self.parse_lines()
307
308     def replace(self, replace_from, replace_to):
309         lines = [] 
310         for l in self.for_writing:
311             lines += [l.replace(replace_from, replace_to)]
312         self.lines = [LedgerTextLine(l) for l in lines]
313         self.clean()
314         self.parse_lines()
315
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
324         past_kk_expenses = 0
325         past_kk_add = 0
326         past_neuanfangspuffer_expenses = 0
327         for b in db.bookings:
328             if b.date == self.date:
329                 break
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['€']
336
337         needed_netto = -self.account_changes['Assets'].money_dict['€']
338         past_taxed_needs_before_kk = past_neuanfangspuffer_expenses - past_kk_expenses
339         ESt_this_month = 0
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
345         too_low = 0
346         too_high = 2 * needed_netto 
347         while True:
348             estimate_for_remaining_year = (12 - months_passed) * taxed_income_before_kk 
349             zvE = past_taxed_needs_before_kk + estimate_for_remaining_year
350             if zvE < E0:
351                 ESt_year = decimal.Decimal(0)
352             elif zvE < E1:
353                 y = (zvE - E0)/10000
354                 ESt_year = (decimal.Decimal(922.98) * y + 1400) * y
355             elif zvE < E2:
356                 y = (zvE - E1)/10000
357                 ESt_year = (decimal.Decimal(181.19) * y + 2397) * y + decimal.Decimal(1025.38)
358             elif zvE < E3:
359                 ESt_year = decimal.Decimal(0.42) * zvE - 10602.13 
360             else:
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:
365                 break
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)]
374
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'
388         else:
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)]
391
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, '€')]
399
400
401 class LedgerDB(PlomDB):
402
403     def __init__(self, prefix, ignore_editable_exceptions=False):
404         self.prefix = prefix 
405         self.bookings = []
406         self.comments = []
407         self.text_lines = []
408         super().__init__(db_path)
409         self.bookings = parse_lines_to_bookings(self.text_lines, ignore_editable_exceptions)
410
411     def read_db_file(self, f):
412         self.text_lines += f.readlines()
413
414     def insert_booking_at_date(self, booking):
415         place_at = 0
416         if len(self.bookings) > 0:
417             for i, iterated_booking in enumerate(self.bookings):
418                 if booking.date < iterated_booking.date:
419                     break
420                 elif booking.date == iterated_booking.date:
421                     place_at = i
422                     break
423                 else:
424                     place_at = i + 1
425         self.bookings.insert(place_at, booking)
426
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)
432
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(':')
440                 path = [] 
441                 for tok in toks:
442                     parent_name = ':'.join(path) 
443                     path += [tok] 
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)
448         class Node:
449             def __init__(self, indent, name, moneys):
450                 self.indent = indent
451                 self.name = name
452                 self.moneys = moneys.money_dict
453         nodes = []
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)
461
462     def edit(self, index, sent=None, error_msg=None, edit_mode='table', copy=False):
463         accounts = set() 
464         if sent or -1 == index:
465             content = sent if sent else ([] if 'textarea'==edit_mode else None)
466         else:
467             content = self.bookings[index]
468         date_today = str(datetime.now())[:10]
469         if copy:
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
475         else:
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))
480
481     def move_up(self, index):
482         return self.move(index, -1) 
483
484     def move_down(self, index):
485         return self.move(index, +1) 
486
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 
493         return swap_index
494
495     def write_db(self):
496         lines = []
497         for i, booking in enumerate(self.bookings):
498             if i > 0:
499                 lines += ['']
500             lines += booking.for_writing
501         self.write_text_to_db('\n'.join(lines) + '\n')
502
503
504 class LedgerHandler(PlomHandler):
505     
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 
511
512     def do_POST(self):
513         self.try_do(self.forward_posts)
514
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)
520         index = 0
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
525                 break
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
534             if 'textarea' == edit_mode:
535                 lines = [LedgerTextLine(line) for line in postvars['booking'][0].rstrip().split('\n')]
536                 booking = Booking(lines, starts_at, validate=validate)
537             else:
538                 booking = Booking.from_postvars(postvars, starts_at, validate)
539             if submit_button in {'update', 'add'}:
540                 if submit_button == 'update':
541                     if 'textarea' == edit_mode and 'delete' == ''.join([l.text_line for l in lines]).strip():
542                        del db.bookings[index]
543                     # if not creating new Booking, and date unchanged, keep it in place 
544                     elif booking.date == db.bookings[index].date:
545                        db.bookings[index] = booking 
546                     else:
547                        del db.bookings[index]
548                        db.insert_booking_at_date(booking)
549                 else: 
550                     db.insert_booking_at_date(booking)
551             else:  # non-DB-writing calls
552                 error_msg = None
553                 if 'check' == submit_button:
554                     error_msg = 'All looks fine!'
555                 elif submit_button in {'mirror', 'fill_sink', 'add_taxes'}:
556                     if 'add_taxes' == submit_button:
557                         booking.add_taxes(db)
558                     else:
559                         getattr(booking, submit_button)()
560                 elif 'replace' == submit_button:
561                     booking.replace(postvars['replace_from'][0], postvars['replace_to'][0])
562                 elif submit_button in {'textarea', 'table'}:
563                     edit_mode = submit_button
564                 page = db.edit(index, booking, error_msg=error_msg, edit_mode=edit_mode)
565                 self.send_HTML(page)
566                 return
567         db.write_db() 
568         index = index if index >= 0 else len(db.bookings) - 1
569         self.redirect(prefix + f'/ledger#{index}')
570
571     def do_GET(self):
572         self.try_do(self.forward_gets)
573
574     def forward_gets(self):
575         prefix = self.apps['ledger'] if hasattr(self, 'apps') else '' 
576         try:
577             db = LedgerDB(prefix=prefix)
578         except EditableException as e:
579             # We catch the EditableException for further editing, and then
580             # re-run the DB initiation without it blocking DB creation.
581             db = LedgerDB(prefix=prefix, ignore_editable_exceptions=True)
582             page = db.edit(index=e.booking_index, error_msg=f'ERROR: {e}')
583             self.send_HTML(page)
584             return
585         parsed_url = urlparse(self.path)
586         params = parse_qs(parsed_url.query)
587         if parsed_url.path == f'{prefix}/balance':
588             stop = params.get('until_after', [None])[0]
589             page = db.balance_as_html(stop)
590         elif parsed_url.path == f'{prefix}/edit':
591             index = params.get('i', [-1])[0]
592             copy = params.get('copy', [0])[0]
593             page = db.edit(int(index), copy=bool(copy))
594         else:
595             page = db.ledger_as_html()
596         self.send_HTML(page)
597
598
599
600 if __name__ == "__main__":  
601     run_server(server_port, LedgerHandler)