X-Git-Url: https://plomlompom.com/repos/berlin_corona.txt?a=blobdiff_plain;f=ledger.py;h=60bbf640a4d117ca548a407d727375cc60dbad1d;hb=6061ff7f1948b83dc601f52df1119a57b6e91649;hp=c3f0ab9c22d028d92b1c2741e7c411a4d9efa499;hpb=2deeead813fbdf0fec4f94d238a43fd93514f94f;p=misc diff --git a/ledger.py b/ledger.py index c3f0ab9..60bbf64 100755 --- a/ledger.py +++ b/ledger.py @@ -1,6 +1,8 @@ from http.server import BaseHTTPRequestHandler, HTTPServer import sys import os +import html +from urllib.parse import parse_qs, urlparse hostName = "localhost" serverPort = 8082 @@ -23,6 +25,104 @@ def apply_booking_to_account_balances(account_sums, account, currency, amount): account_sums[account][currency] += amount +def add_taxes(lines): + import decimal + bookings, _ = parse_lines(lines.copy()) + _, account_sums = bookings_to_account_tree(bookings) + expenses_so_far = -1 * account_sums['Assets']['€'] + needed_income_before_krankenkasse = expenses_so_far + ESt_this_month = 0 + left_over = needed_income_before_krankenkasse - ESt_this_month + too_low = 0 + too_high = 2 * needed_income_before_krankenkasse + E0 = 10908 + E1 = 15999 + E2 = 62809 + E3 = 277825 + while True: + zvE = 12 * needed_income_before_krankenkasse + if zvE < E0: + ESt = decimal.Decimal(0) + elif zvE < E1: + y = (zvE - E0)/10000 + ESt = (decimal.Decimal(979.18) * y + 1400) * y + elif zvE < E2: + y = (zvE - E1)/10000 + ESt = (decimal.Decimal(192.59) * y + 2397) * y + decimal.Decimal(966.53) + elif zvE < E3: + ESt = decimal.Decimal(0.42) * (zvE - decimal.Decimal(62809)) + decimal.Decimal(16405.54) + else: + ESt = decimal.Decimal(0.45) * (zvE - decimal.Decimal(277825)) + decimal.Decimal(106713.52) + ESt_this_month = ESt / 12 + left_over = needed_income_before_krankenkasse - ESt_this_month + if abs(left_over - expenses_so_far) < 0.1: + break + elif left_over < expenses_so_far: + too_low = needed_income_before_krankenkasse + elif left_over > expenses_so_far: + too_high = needed_income_before_krankenkasse + needed_income_before_krankenkasse = too_low + (too_high - too_low)/2 + line_income_tax = f' Reserves:Einkommenssteuer {ESt_this_month:.2f}€ ; expenses so far: {expenses_so_far:.2f}€; zvE: {zvE:.2f}€; ESt total: {ESt:.2f}€; needed before Krankenkasse: {needed_income_before_krankenkasse:.2f}€' + kk_minimum_income = 1096.67 + kk_factor = decimal.Decimal(0.189) + kk_minimum_tax = decimal.Decimal(207.27) + # kk_minimum_income = 1131.67 + # kk_factor = decimal.Decimal(0.191) + # kk_minimum_tax = decimal.Decimal(216.15) + # kk_factor = decimal.Decimal(0.197) + # kk_minimum_tax = decimal.Decimal(222.94) + kk_add = max(0, kk_factor * needed_income_before_krankenkasse - kk_minimum_tax) + line_kk_minimum = f' Reserves:Month:Krankenkassendefaultbeitrag {kk_minimum_tax:.2f}€ ; assed minimum income {kk_minimum_income:.2f}€ * {kk_factor:.3f}' + line_kk_add = f' Reserves:Month:Krankenkassenbeitragswachstum {kk_add:.2f}€ ; max(0, {kk_factor:.3f} * {needed_income_before_krankenkasse:.2f}€ - {kk_minimum_tax:.2f}€)' + final_minus = expenses_so_far + ESt_this_month + kk_minimum_tax + kk_add + line_finish = f' Assets -{final_minus:.2f}€' + return [line_income_tax, line_kk_minimum, line_kk_add, line_finish] + + +def bookings_to_account_tree(bookings): + account_sums = {} + for booking in bookings: + for account, changes in booking.account_changes.items(): + for currency, amount in changes.items(): + apply_booking_to_account_balances(account_sums, account, currency, amount) + account_tree = {} + def collect_branches(account_name, path): + node = account_tree + path_copy = path[:] + while len(path_copy) > 0: + step = path_copy.pop(0) + node = node[step] + toks = account_name.split(":", maxsplit=1) + parent = toks[0] + if parent in node.keys(): + child = node[parent] + else: + child = {} + node[parent] = child + if len(toks) == 2: + k, v = collect_branches(toks[1], path + [parent]) + if k not in child.keys(): + child[k] = v + else: + child[k].update(v) + return parent, child + for account_name in sorted(account_sums.keys()): + k, v = collect_branches(account_name, []) + if k not in account_tree.keys(): + account_tree[k] = v + else: + account_tree[k].update(v) + def collect_totals(parent_path, tree_node): + for k, v in tree_node.items(): + child_path = parent_path + ":" + k + for currency, amount in collect_totals(child_path, v).items(): + apply_booking_to_account_balances(account_sums, parent_path, currency, amount) + return account_sums[parent_path] + for account_name in account_tree.keys(): + account_sums[account_name] = collect_totals(account_name, account_tree[account_name]) + return account_tree, account_sums + + def parse_lines(lines): import datetime import decimal @@ -32,6 +132,7 @@ def parse_lines(lines): start_line = 0 bookings = [] comments = [] + lines += [''] # to ensure a booking-ending last line for i, line in enumerate(lines): prefix = f"line {i}" # we start with the case of an utterly empty line @@ -52,7 +153,7 @@ def parse_lines(lines): # if non-empty line, first get comment if any, and commit to DB split_by_comment = stripped_line.split(sep=";", maxsplit=1) if len(split_by_comment) == 2: - comments[i] = split_by_comment[1] + comments[i] = split_by_comment[1].lstrip() # if pre-comment empty: if inside_booking, this must be a comment-only line so we keep it for later ledger-output to capture those comments; otherwise, no more to process for this line non_comment = split_by_comment[0].rstrip() if non_comment.rstrip() == '': @@ -73,6 +174,7 @@ def parse_lines(lines): except IndexError: raise HandledException(f"{prefix} bad description: {description}") inside_booking = True + booking_lines += [non_comment] continue # otherwise, read as transfer data toks = non_comment.split() # ignore specification's allowance of single spaces in names @@ -156,7 +258,7 @@ class Booking: prefix = f"booking at line {self.start_line}" sums = {} empty_values = 0 - for line in self.lines: + for line in self.lines[1:]: if line == '': continue _, amount, currency = line @@ -184,7 +286,7 @@ class Booking: account_changes = {} debt = {} sink_account = None - for line in self.lines: + for line in self.lines[1:]: if line == '': continue account, amount, currency = line @@ -205,17 +307,35 @@ class Booking: class Database: - def __init__(self, load_from_file=True): + def __init__(self): db_name = "_ledger" self.db_file = db_name + ".json" self.lock_file = db_name+ ".lock" self.bookings = [] self.comments = [] - if load_from_file and os.path.exists(self.db_file): + self.real_lines = [] + if os.path.exists(self.db_file): with open(self.db_file, "r") as f: - ret = parse_lines(f.readlines()) - self.bookings += ret[0] - self.comments += ret[1] + self.real_lines += f.readlines() + ret = parse_lines(self.real_lines) + self.bookings += ret[0] + self.comments += ret[1] + + def get_lines(self, start, end): + return self.real_lines[start:end] + + def replace(self, start, end, lines): + import shutil + if os.path.exists(self.lock_file): + raise HandledException('Sorry, lock file!') + if os.path.exists(self.db_file): + shutil.copy(self.db_file, self.db_file + ".bak") + f = open(self.lock_file, 'w+') + f.close() + text = ''.join(self.real_lines[:start]) + '\n'.join(lines) + ''.join(self.real_lines[end:]) + with open(self.db_file, 'w') as f: + f.write(text); + os.remove(self.lock_file) def append(self, lines): import shutil @@ -231,25 +351,72 @@ class Database: class MyServer(BaseHTTPRequestHandler): - header = 'ledger balance add
' - footer = '' + header = """ + + + +ledger +balance +add free +add structured +
+""" + footer = "\n" def do_POST(self): - from urllib.parse import parse_qs + db = Database() length = int(self.headers['content-length']) postvars = parse_qs(self.rfile.read(length).decode(), keep_blank_values=1) - lines = postvars['booking'][0].splitlines() + parsed_url = urlparse(self.path) + if '/add_structured' == parsed_url.path: + date = postvars['date'][0] + description = postvars['description'][0] + start_comment = postvars['line_0_comment'][0] + lines = [f'{date} {description} ; {start_comment}'] + i = 1 + while f'line_{i}_comment' in postvars.keys(): + account = postvars[f'line_{i}_account'][0] + amount = postvars[f'line_{i}_amount'][0] + currency = postvars[f'line_{i}_currency'][0] + comment = postvars[f'line_{i}_comment'][0] + i += 1 + new_main = f'{account} {amount} {currency}' + if '' == new_main.rstrip() == comment.rstrip(): # don't write empty lines + continue + new_line = new_main + if comment.rstrip() != '': + new_line += f' ; {comment}' + lines += [new_line] + if 'add_income_tax' in postvars.keys(): + lines += add_taxes(lines) + elif '/add_free' == parsed_url.path: + lines = postvars['booking'][0].splitlines() + start = int(postvars['start'][0]) + end = int(postvars['end'][0]) try: - bookings, comments = parse_lines(lines) - db.append(lines) - self.send_response(200) + _, _ = parse_lines(lines) + if start == end == 0: + db.append(lines) + else: + db.replace(start, end, lines) + self.send_response(301) + redir_url = '/' + if 'add_income_tax' in postvars.keys(): + redir_url = f'/add_structured?start={start}&end={start+len(lines)-1}' + self.send_header('Location', redir_url) self.end_headers() - page = f"{self.header}Success!{self.footer}" - self.wfile.write(bytes(page, "utf-8")) except HandledException as e: self.send_response(400) self.end_headers() - page = f"{self.header}{e}{self.footer}" + page = f'{self.header}ERROR: {e}{self.footer}' self.wfile.write(bytes(page, "utf-8")) def do_GET(self): @@ -257,58 +424,30 @@ class MyServer(BaseHTTPRequestHandler): self.send_header("Content-type", "text/html") self.end_headers() db = Database() + parsed_url = urlparse(self.path) page = self.header + '' - if self.path == '/balance': + params = parse_qs(parsed_url.query) + start = int(params.get('start', ['0'])[0]) + end = int(params.get('end', ['0'])[0]) + bonus_lines = int(params.get('bonus_lines', ['0'])[0]) + if parsed_url.path == '/balance': page += self.balance_as_html(db) - elif self.path == '/add': - page += self.add() + elif parsed_url.path == '/add_free': + page += self.add_free(db, start, end) + elif parsed_url.path == '/add_structured': + page += self.add_structured(db, start, end) + elif parsed_url.path == '/copy_free': + page += self.add_free(db, start, end, copy=True) + elif parsed_url.path == '/copy_structured': + page += self.add_structured(db, start, end, copy=True) else: page += self.ledger_as_html(db) page += self.footer self.wfile.write(bytes(page, "utf-8")) def balance_as_html(self, db): - account_sums = {} - for booking in db.bookings: - for account, changes in booking.account_changes.items(): - for currency, amount in changes.items(): - apply_booking_to_account_balances(account_sums, account, currency, amount) - account_tree = {} - def collect_branches(account_name, path): - node = account_tree - path_copy = path[:] - while len(path_copy) > 0: - step = path_copy.pop(0) - node = node[step] - toks = account_name.split(":", maxsplit=1) - parent = toks[0] - if parent in node.keys(): - child = node[parent] - else: - child = {} - node[parent] = child - if len(toks) == 2: - k, v = collect_branches(toks[1], path + [parent]) - if k not in child.keys(): - child[k] = v - else: - child[k].update(v) - return parent, child - for account_name in sorted(account_sums.keys()): - k, v = collect_branches(account_name, []) - if k not in account_tree.keys(): - account_tree[k] = v - else: - account_tree[k].update(v) - def collect_totals(parent_path, tree_node): - for k, v in tree_node.items(): - child_path = parent_path + ":" + k - for currency, amount in collect_totals(child_path, v).items(): - apply_booking_to_account_balances(account_sums, parent_path, currency, amount) - return account_sums[parent_path] - for account_name in account_tree.keys(): - account_sums[account_name] = collect_totals(account_name, account_tree[account_name]) lines = [] + account_tree, account_sums = bookings_to_account_tree(db.bookings) def print_subtree(lines, indent, node, subtree, path): line = f"{indent}{node}" n_tabs = 5 - (len(line) // 8) @@ -333,27 +472,108 @@ class MyServer(BaseHTTPRequestHandler): def ledger_as_html(self, db): lines = [] for comment in db.comments: - lines += [f"; {comment}" if comment != '' else ''] + lines += [f'{comment}' if comment != '' else ''] for booking in db.bookings: i = booking.start_line - suffix = f" {lines[i]}" if len(lines[i]) > 0 else "" - lines[i] = f"{booking.date_string} {booking.description}{suffix}" - for booking_line in booking.lines: + booking_end = booking.start_line + len(booking.lines) + lines[i] = f"""{booking.date_string} {booking.description} {lines[i]} +[edit: structured +/ free +| copy:structured +/ free] +""" + for booking_line in booking.lines[1:]: i += 1 if booking_line == '': + lines[i] = f'' continue - suffix = f" {lines[i]}" if len(lines[i]) > 0 else "" - value = f" {booking_line[1]} {booking_line[2]}" if booking_line[1] else "" - lines[i] = f"{booking_line[0]}{value}{suffix}" - content = "\n".join(lines) - return f"
{content}
" + comment = f' {lines[i]}' if len(lines[i]) > 0 else '' + money = '' + if booking_line[1]: + money = f'{booking_line[1]} {booking_line[2]}' + account = booking_line[0] + lines[i] = f'' + lines[i] = lines[i] + "\n" + '
{lines[i]}
{booking_line[0]}{money}{comment}
' + return '\n'.join(lines) + + def header_add_form(self, action): + return f"
\n" - def add(self): - return '
' + def footer_add_form(self, start, end, copy): + if copy: + start = end = 0 + return f""" + + + +""" + + def textarea(self, name, lines, min_rows=1, min_cols=80): + safe_content = html.escape(''.join(lines)) + n_rows = max(min_rows, len(lines)) + n_cols = min_cols + for line in lines: + n_cols = len(line) if len(line) > n_cols else n_cols + return f'' + + def add_free(self, db, start=0, end=0, copy=False): + lines = db.get_lines(start, end) + return f'{self.header_add_form("add_free")}{self.textarea("booking",lines,10)}{self.footer_add_form(start, end, copy)}' + + def add_structured(self, db, start=0, end=0, bonus_lines=10, copy=False): + import datetime + lines = db.get_lines(start, end) + bookings, comments = parse_lines(lines) + if len(bookings) > 1: + raise HandledException('can only edit single Booking') + last_line = 0 + def inpu(name, val="", datalist="", input_type='text', size=-1): + val = val if val is not None else "" + safe_val = html.escape(str(val)) + datalist_string = '' if datalist == '' else f'list="{datalist}"' + number_step = '' if input_type != 'number' else ' step=0.01' + size_string = '' if size < 0 else f' size={size}' + return f'' + input_lines = inpu('add_income_tax', 'add income tax', '', 'submit') + '
' + today = str(datetime.datetime.now())[:10] + if len(bookings) == 0: + input_lines += f'{inpu("date", today, size=9)} {inpu("description", "", "descriptions")} ; {inpu("line_0_comment")}
' + last_line = 1 + else: + booking = bookings[0] + last_line = len(comments) + date_string = today if copy else booking.date_string + input_lines += f'{inpu("date", date_string, size=9)} {inpu("description", booking.description, "descriptions")} ; {self.textarea("line_0_comment", [comments[0]])}
' + for i in range(1, len(comments)): + account = amount = currency = '' + if i < len(booking.lines) and booking.lines[i] != '': + account = booking.lines[i][0] + amount = booking.lines[i][1] + currency = booking.lines[i][2] + input_lines += f'{inpu(f"line_{i}_account", account, "accounts", size=40)} {inpu(f"line_{i}_amount", amount, "", "number", size=10)} {inpu(f"line_{i}_currency", currency, "currencies", size=3)} ; {self.textarea(f"line_{i}_comment", [comments[i]])}
' + for j in range(bonus_lines): + i = j + last_line + input_lines += f'{inpu(f"line_{i}_account", "", "accounts", size=40)} {inpu(f"line_{i}_amount", "", "", "number", size=10)} {inpu(f"line_{i}_currency", "", "currencies", size=3)} ; {self.textarea(f"line_{i}_comment", [""])}
' + datalist_sets = {'descriptions': set(), 'accounts': set(), 'currencies': set()} + for b in db.bookings: + datalist_sets['descriptions'].add(b.description) + for account, moneys in b.account_changes.items(): + datalist_sets['accounts'].add(account) + for currency in moneys.keys(): + datalist_sets['currencies'].add(currency) + def build_datalist(name): + datalist = f'' + "\n" + for item in datalist_sets[name]: + safe_item = html.escape(item) + datalist += f'' + "\n" + return f"{datalist}\n" + datalists = build_datalist('descriptions') + datalists += build_datalist('accounts') + datalists += build_datalist('currencies') + return f'{self.header_add_form("add_structured")}{input_lines}{datalists}{self.footer_add_form(start, end, copy)}' -db = Database() -if __name__ == "__main__": +if __name__ == "__main__": webServer = HTTPServer((hostName, serverPort), MyServer) print(f"Server started http://{hostName}:{serverPort}") try: