home · contact · privacy
Add forgotten files.
[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 from os.path import split as path_split
8
9 db_path = '/home/plom/org/ledger2024.dat'
10 server_port = 8082
11 j2env = JinjaEnv(loader=JinjaFSLoader('ledger_templates'))
12
13 class EditableException(PlomException):
14     def __init__(self, booking_index, *args, **kwargs):
15         self.booking_index = booking_index
16         super().__init__(*args, **kwargs)
17     
18
19 class LedgerTextLine:
20
21     def __init__(self, text_line):
22         self.text_line = text_line
23         self.comment = '' 
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
27         if self.empty:
28             return
29         if len(split_by_comment) == 2:
30             self.comment = split_by_comment[1].lstrip()
31
32
33 class Wealth:
34
35     def __init__(self):
36         self.money_dict = {}
37
38     def __iadd__(self, moneys):
39         money_dict = 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
46         return self
47
48     @property
49     def sink_empty(self):
50         return len(self.as_sink) == 0
51
52     @property
53     def as_sink(self):
54         sink = {} 
55         for currency, amount in self.money_dict.items():
56             if 0 == amount:
57                 continue
58             sink[currency] = -amount
59         return sink 
60
61
62 class Account:
63
64     def __init__(self, own_name, parent):
65         self.own_name = own_name
66         self.parent = parent
67         if self.parent:
68             self.parent.children += [self]
69         self.own_moneys = Wealth() 
70         self.children = []
71
72     def add_wealth(self, moneys):
73         self.own_moneys += moneys
74
75     @property
76     def full_name(self):
77         if self.parent and len(self.parent.own_name) > 0:
78             return f'{self.parent.full_name}:{self.own_name}'
79         else:
80             return self.own_name
81
82     @property
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
88         return full_moneys
89
90
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
94     bookings = []
95     inside_booking = False
96     booking_lines = []
97     booking_start_i = 0
98     last_date = '' 
99     for i, line in enumerate(lines):
100         intro = f'file line {i}'
101         if line.empty: 
102             if inside_booking:
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
109                 booking_lines = []
110         else:
111             if not inside_booking:
112                 booking_start_i = i
113             inside_booking = True
114             booking_lines += [line]
115     if inside_booking:
116         raise PlomException(f'{intro}: last booking unfinished')
117     return bookings
118
119
120 class TransferLine:
121
122     def __init__(self, line=None, account='', amount=None, currency='', comment='', validate=True):
123         self.account = account 
124         self.amount = amount 
125         self.currency = currency 
126         self.comment = comment 
127         if line:
128             if line.empty:
129                 raise PlomException('line empty')
130             self.comment = line.comment
131             toks = line.non_comment.split()
132             if (len(toks) not in {1, 3}):
133                 if validate:
134                     raise PlomException(f'number of non-comment tokens not 1 or 3')
135                 elif len(toks) == 2:
136                     toks += ['']
137                 else:
138                     toks = 3*['']
139             self.account = toks[0]
140             if len(toks) != 1:
141                 try:
142                     self.amount = decimal.Decimal(toks[1])
143                 except decimal.InvalidOperation:
144                     if validate:
145                         raise PlomException(f'invalid token for Decimal: {toks[1]}')
146                     else:
147                         self.comment = f'unparsed: {toks[1]}; {self.comment}'
148                 self.currency = toks[2]
149
150     @property
151     def amount_fmt(self):
152         if self.amount is None:
153             return ''
154         elif self.amount.as_tuple().exponent < -2:
155             return f'{self.amount:.1f}…'
156         else:
157             return f'{self.amount:.2f}'
158
159     @property
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}' 
166         return line
167
168     @property
169     def comment_cols(self):
170         return max(20, len(self.comment))
171
172
173 class Booking:
174
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}'
179         self.clean()
180         if lines:
181             self.lines = lines 
182             self.parse_lines()
183         else:
184             self.date = date
185             self.description = description
186             self.top_comment = top_comment 
187             self.validate_head()
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]
193
194     @classmethod
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] 
199         transfer_lines = []
200         for i, account in enumerate(postvars['account']):
201             if len(account) == 0:
202                 continue
203             amount = None
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)
208
209     def clean(self):
210         self.transfer_lines = []
211         self.account_changes = {}
212
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)
221         if len(toks) < 2:
222             if self.validate:
223                 raise PlomException(f'{self.intro}: headline missing elements: {non_comment}')
224             elif 0 == len(toks):
225                 toks = 2*['']
226             else:
227                 toks += ['']
228         self.date = toks[0]
229         self.description = toks[1]
230         self.validate_head()
231         for i, line in enumerate(self.lines[1:]):
232             try:
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()
237
238     def calculate_account_changes(self):
239         sink_account = None 
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
247             else:
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
259
260     @property
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]
267         return lines
268
269     @property
270     def comment_cols(self):
271         return max(20, len(self.top_comment))
272
273     def validate_head(self):
274         if not self.validate:
275             return
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')
280         try:
281             datetime.strptime(self.date, '%Y-%m-%d')
282         except ValueError:
283             raise PlomException(f'{self.intro}: bad headline date format: {self.date}')
284
285     def fill_sink(self):
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]
291                 break
292         lines = self.lines[:i+1] + [LedgerTextLine(l) for l in replacement_lines] + self.lines[i+2:]
293         self.clean()
294         self.lines = lines
295         self.parse_lines()
296
297     def mirror(self):
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()}'
302             new_account = '?'
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)]
306         self.clean()
307         self.parse_lines()
308
309     def replace(self, replace_from, replace_to):
310         lines = [] 
311         for l in self.for_writing:
312             lines += [l.replace(replace_from, replace_to)]
313         self.lines = [LedgerTextLine(l) for l in lines]
314         self.clean()
315         self.parse_lines()
316
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
325         past_kk_expenses = 0
326         past_kk_add = 0
327         past_neuanfangspuffer_expenses = 0
328         for b in db.bookings:
329             if b.date == self.date:
330                 break
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['€']
337
338         needed_netto = -self.account_changes['Assets'].money_dict['€']
339         past_taxed_needs_before_kk = past_neuanfangspuffer_expenses - past_kk_expenses
340         ESt_this_month = 0
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
346         too_low = 0
347         too_high = 2 * needed_netto 
348         while True:
349             estimate_for_remaining_year = (12 - months_passed) * taxed_income_before_kk 
350             zvE = past_taxed_needs_before_kk + estimate_for_remaining_year
351             if zvE < E0:
352                 ESt_year = decimal.Decimal(0)
353             elif zvE < E1:
354                 y = (zvE - E0)/10000
355                 ESt_year = (decimal.Decimal(922.98) * y + 1400) * y
356             elif zvE < E2:
357                 y = (zvE - E1)/10000
358                 ESt_year = (decimal.Decimal(181.19) * y + 2397) * y + decimal.Decimal(1025.38)
359             elif zvE < E3:
360                 ESt_year = decimal.Decimal(0.42) * zvE - 10602.13 
361             else:
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:
366                 break
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)]
375
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'
389         else:
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)]
392
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, '€')]
400
401
402 class LedgerDB(PlomDB):
403
404     def __init__(self, prefix, ignore_editable_exceptions=False):
405         self.prefix = prefix 
406         self.bookings = []
407         self.comments = []
408         self.text_lines = []
409         super().__init__(db_path)
410         self.bookings = parse_lines_to_bookings(self.text_lines, ignore_editable_exceptions)
411
412     def read_db_file(self, f):
413         self.text_lines += f.readlines()
414
415     def insert_booking_at_date(self, booking):
416         place_at = 0
417         if len(self.bookings) > 0:
418             for i, iterated_booking in enumerate(self.bookings):
419                 if booking.date < iterated_booking.date:
420                     break
421                 elif booking.date == iterated_booking.date:
422                     place_at = i
423                     break
424                 else:
425                     place_at = i + 1
426         self.bookings.insert(place_at, booking)
427
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)
433
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(':')
441                 path = [] 
442                 for tok in toks:
443                     parent_name = ':'.join(path) 
444                     path += [tok] 
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)
449         class Node:
450             def __init__(self, indent, name, moneys):
451                 self.indent = indent
452                 self.name = name
453                 self.moneys = moneys.money_dict
454         nodes = []
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)
462
463     def edit(self, index, sent=None, error_msg=None, edit_mode='table', copy=False):
464         accounts = set() 
465         if sent or -1 == index:
466             content = sent if sent else ([] if 'textarea'==edit_mode else None)
467         else:
468             content = self.bookings[index]
469         date_today = str(datetime.now())[:10]
470         if copy:
471             content.date = date_today 
472             index = -1
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
477         else:
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(
482                 content=content,
483                 index=index,
484                 error_msg=error_msg,
485                 edit_mode=edit_mode,
486                 accounts=accounts,
487                 adding=-1 == index)
488
489     def move_up(self, index):
490         return self.move(index, -1) 
491
492     def move_down(self, index):
493         return self.move(index, +1) 
494
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 
501         return swap_index
502
503     def write_db(self):
504         lines = []
505         for i, booking in enumerate(self.bookings):
506             if i > 0:
507                 lines += ['']
508             lines += booking.for_writing
509         self.write_text_to_db('\n'.join(lines) + '\n')
510
511
512 class LedgerHandler(PlomHandler):
513     
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 
519
520     def do_POST(self):
521         self.try_do(self.forward_posts)
522
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)
528         index = 0
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
534                 break
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]))
538         elif 'edit' == site:
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
543             delete = False
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':
547                     delete = True
548                 else:
549                     booking = Booking(lines, starts_at, validate=validate)
550             else:
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 
559                     else:
560                        del db.bookings[index]
561                        db.insert_booking_at_date(booking)
562                 else: 
563                     db.insert_booking_at_date(booking)
564             else:  # non-DB-writing calls
565                 error_msg = None
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)
571                     else:
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)
578                 self.send_HTML(page)
579                 return
580         db.write_db() 
581         index = index if index >= 0 else len(db.bookings) - 1
582         self.redirect(f'ledger#{index}')
583
584     def do_GET(self):
585         self.try_do(self.forward_gets)
586
587     def forward_gets(self):
588         prefix = self.apps['ledger'] if hasattr(self, 'apps') else '' 
589         try:
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}')
596             self.send_HTML(page)
597             return
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)
604         elif site == 'edit':
605             index = params.get('i', [-1])[0]
606             copy = params.get('copy', [0])[0]
607             page = db.edit(int(index), copy=bool(copy))
608         else:
609             page = db.ledger_as_html()
610         self.send_HTML(page)
611
612
613
614 if __name__ == "__main__":  
615     run_server(server_port, LedgerHandler)