X-Git-Url: https://plomlompom.com/repos/do_day?a=blobdiff_plain;f=ledger.py;h=f29f73ec190d3b1391b4a3f05ca088607ac221d2;hb=76210b693b77e67bf8160a7fb869ae416309d333;hp=334318ff301eb186efa6684809c9ebac1807d177;hpb=c3b5860ae1ca76c2a1dc3b61d3c1242d581424f0;p=misc diff --git a/ledger.py b/ledger.py index 334318f..f29f73e 100755 --- a/ledger.py +++ b/ledger.py @@ -1,316 +1,1371 @@ -from http.server import BaseHTTPRequestHandler, HTTPServer -import sys -hostName = "localhost" -serverPort = 8082 +import os +import decimal +from datetime import datetime, timedelta +from urllib.parse import parse_qs, urlparse +from jinja2 import Environment as JinjaEnv, FileSystemLoader as JinjaFSLoader +from plomlib import PlomDB, PlomException, run_server, PlomHandler +db_path = '/home/plom/org/ledger2024.dat' +server_port = 8082 +j2env = JinjaEnv(loader=JinjaFSLoader('ledger_templates')) -class HandledException(Exception): - pass +class EditableException(PlomException): + def __init__(self, booking_index, *args, **kwargs): + self.booking_index = booking_index + super().__init__(*args, **kwargs) + +class LedgerTextLine: -def handled_error_exit(msg): - print(f"ERROR: {msg}") - sys.exit(1) + def __init__(self, text_line): + self.text_line = text_line + self.comment = '' + split_by_comment = text_line.rstrip().split(sep=';', maxsplit=1) + self.non_comment = split_by_comment[0].rstrip() + self.empty = len(split_by_comment) == 1 and len(self.non_comment) == 0 + if self.empty: + return + if len(split_by_comment) == 2: + self.comment = split_by_comment[1].lstrip() -def apply_booking_to_account_balances(account_sums, account, currency, amount): - if not account in account_sums: - account_sums[account] = {currency: amount} - elif not currency in account_sums[account].keys(): - account_sums[account][currency] = amount - else: - account_sums[account][currency] += amount +# html_head = """ +# +#
+# ledger +# ledger2 +# balance +# balance2 +# add free +# add structured +# add +#{{date}} {{desc}} {{head_comment|e}}
+#
+#
{{l.acc|e}} | {{l.money|e}} | {{l.comment|e}} |
{{l.comment|e}} |
{content}" - - def balance_as_html(self): - account_sums = {} - for booking in self.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] + @property + def amount_fmt(self): + if self.amount is None: + return '' + elif self.amount.as_tuple().exponent < -2: + return f'{self.amount:.1f}â¦' + else: + return f'{self.amount:.2f}' + + @property + def for_writing(self): + line = f' {self.account}' + if self.amount is not None: + line += f' {self.amount} {self.currency}' + if self.comment != '': + line += f' ; {self.comment}' + return line + + @property + def comment_cols(self): + return max(20, len(self.comment)) + + +class Booking: + + def __init__(self, lines=None, starts_at='?', date='', description='', top_comment='', transfer_lines=None, validate=True): + self.validate = validate + self.starts_at = starts_at + self.intro = f'booking starting at line {self.starts_at}' + self.clean() + if lines: + self.lines = lines + self.parse_lines() + else: + self.date = date + self.description = description + self.top_comment = top_comment + self.validate_head() + self.transfer_lines = transfer_lines if transfer_lines else [] + if self.validate and len(self.transfer_lines) < 2: + raise PlomException(f'{self.intro}: too few transfer lines') + self.calculate_account_changes() + self.lines = [LedgerTextLine(l) for l in self.for_writing] + + @classmethod + def from_postvars(cls, postvars, starts_at='?', validate=True): + date = postvars['date'][0] + description = postvars['description'][0] + top_comment = postvars['top_comment'][0] + transfer_lines = [] + for i, account in enumerate(postvars['account']): + if len(account) == 0: + continue + amount = None + if len(postvars['amount'][i]) > 0: + amount = decimal.Decimal(postvars['amount'][i]) + transfer_lines += [TransferLine(None, account, amount, postvars['currency'][i], postvars['comment'][i])] + return cls(None, starts_at, date, description, top_comment, transfer_lines, validate=validate) + + def clean(self): + self.transfer_lines = [] + self.account_changes = {} + + def parse_lines(self): + if len(self.lines) < 3 and self.validate: + raise PlomException(f'{self.intro}: ends with less than 3 lines:' + str(self.lines)) + top_line = self.lines[0] + if top_line.empty and self.validate: + raise PlomException('{self.intro}: headline empty') + self.top_comment = top_line.comment + toks = top_line.non_comment.split(maxsplit=1) + if len(toks) < 2: + if self.validate: + raise PlomException(f'{self.intro}: headline missing elements: {non_comment}') + elif 0 == len(toks): + toks = 2*[''] + else: + toks += [''] + self.date = toks[0] + self.description = toks[1] + self.validate_head() + for i, line in enumerate(self.lines[1:]): + try: + self.transfer_lines += [TransferLine(line, validate=self.validate)] + except PlomException as e: + raise PlomException(f'{self.intro}, transfer line {i}: {e}') + self.calculate_account_changes() + + def calculate_account_changes(self): + sink_account = None + money_changes = Wealth() + for i, transfer_line in enumerate(self.transfer_lines): + intro = f'{self.intro}, transfer line {i}' + if transfer_line.amount is None: + if sink_account is not None and self.validate: + raise PlomException(f'{intro}: second sink found (only one allowed)') + sink_account = transfer_line.account + else: + if not transfer_line.account in self.account_changes.keys(): + self.account_changes[transfer_line.account] = Wealth() + money = {transfer_line.currency: transfer_line.amount} + self.account_changes[transfer_line.account] += money + money_changes += money + if sink_account is None and (not money_changes.sink_empty) and self.validate: + raise PlomException(f'{intro}: does not balance (undeclared non-empty sink)') + if sink_account is not None: + if not sink_account in self.account_changes.keys(): + self.account_changes[sink_account] = Wealth() + self.account_changes[sink_account] += money_changes.as_sink + + @property + def for_writing(self): + lines = [f'{self.date} {self.description}'] + if self.top_comment is not None and self.top_comment.rstrip() != '': + lines[0] += f' ; {self.top_comment}' + for line in self.transfer_lines: + lines += [line.for_writing] + return lines + + @property + def comment_cols(self): + return max(20, len(self.top_comment)) + + def validate_head(self): + if not self.validate: + return + if len(self.date) == 0: + raise PlomException(f'{self.intro}: missing date') + if len(self.description) == 0: + raise PlomException(f'{self.intro}: missing description') + try: + datetime.strptime(self.date, '%Y-%m-%d') + except ValueError: + raise PlomException(f'{self.intro}: bad headline date format: {self.date}') + + def fill_sink(self): + replacement_lines = [] + for i, line in enumerate(self.transfer_lines): + if line.amount is None: + for currency, amount in self.account_changes[line.account].money_dict.items(): + replacement_lines += [TransferLine(None, f'{line.account}', amount, currency).for_writing] + break + lines = self.lines[:i+1] + [LedgerTextLine(l) for l in replacement_lines] + self.lines[i+2:] + self.clean() + self.lines = lines + self.parse_lines() + + def mirror(self): + new_transfer_lines = [] + for transfer_line in self.transfer_lines: + uncommented_source = LedgerTextLine(transfer_line.for_writing).non_comment + comment = f're: {uncommented_source.lstrip()}' + new_account = '?' + new_transfer_lines += [TransferLine(None, new_account, -transfer_line.amount, transfer_line.currency, comment)] + for transfer_line in new_transfer_lines: + self.lines += [LedgerTextLine(transfer_line.for_writing)] + self.clean() + self.parse_lines() + + def replace(self, replace_from, replace_to): + lines = [] + for l in self.for_writing: + lines += [l.replace(replace_from, replace_to)] + self.lines = [LedgerTextLine(l) for l in lines] + self.clean() + self.parse_lines() + + def add_taxes(self): + acc_kk_add = 'Reserves:KrankenkassenBeitragsWachstum' + acc_kk_minimum = 'Reserves:Month:KrankenkassenDefaultBeitrag' + acc_kk = 'Expenses:KrankenKasse' + acc_est = 'Reserves:EinkommensSteuer' + acc_assets = 'Assets' + acc_buffer = 'Reserves:NeuAnfangsPuffer:Ausgaben' + buffer_expenses = 0 # FIXME: hardcoded for now + kk_expenses = 0 # FIXME: hardcoded for now + est_expenses = 0 # FIXME: hardcoded for now + months_passed = 0 # FIXME: hardcoded for now + last_monthbreak_assets = 0 # FIXME: hardcoded for now + last_monthbreak_est = 0 # FIXME: hardcoded for now + last_monthbreak_kk_minimum = 0 # FIXME: hardcoded for now + last_monthbreak_kk_add = 0 # FIXME: hardcoded for now + expenses_so_far = 0 # FIXME: hardcoded for now + needed_netto = self.account_changes['Assets'].money_dict['â¬'] + ESt_this_month = 0 + left_over = needed_netto - ESt_this_month + too_low = 0 + too_high = 2 * needed_netto + E0 = decimal.Decimal(10908) + E1 = decimal.Decimal(15999) + E2 = decimal.Decimal(62809) + E3 = decimal.Decimal(277825) + while True: + zvE = buffer_expenses - kk_expenses + (12 - months_passed) * needed_netto + # if finish: + # zvE += last_monthbreak_assets + last_monthbreak_kk_add + last_monthbreak_kk_minimum + 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: - 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 + ESt = decimal.Decimal(0.45) * (zvE - decimal.Decimal(277825)) + decimal.Decimal(106713.52) + ESt_this_month = (ESt + last_monthbreak_est - est_expenses) / (12 - months_passed) + left_over = needed_netto - ESt_this_month + if abs(left_over - expenses_so_far) < 0.001: + break + elif left_over < expenses_so_far: + too_low = needed_netto + elif left_over > expenses_so_far: + too_high = needed_netto + needed_netto = too_low + (too_high - too_low)/2 + ESt_this_month = ESt_this_month.quantize(decimal.Decimal('0.00')) + self.transfer_lines += [TransferLine(None, acc_est, ESt_this_month, 'â¬')] + kk_minimum_income = 1131.67 + kk_factor = decimal.Decimal(0.197) + kk_minimum_tax = decimal.Decimal(222.94).quantize(decimal.Decimal('0.00')) + kk_add_so_far = account_sums[acc_kk_add]['â¬'] if acc_kk_add in account_sums.keys() else 0 + kk_add = needed_netto / (1 - kk_factor) - needed_netto - kk_minimum_tax + hit_kk_minimum_income_limit = False + if kk_add_so_far + kk_add < 0: + hit_kk_minimum_income_limit = True + kk_add_uncorrect = kk_add + kk_add = -(kk_add + kk_add_so_far) + kk_add = decimal.Decimal(kk_add).quantize(decimal.Decimal('0.00')) + self.transfer_lines += [TransferLine(None, acc_kk_minimum, kk_minimum_tx, 'â¬')] + self.transfer_lines += [TransferLine(None, acc_kk_add, kk_add, 'â¬')] + diff = - last_monthbreak_est + ESt_this_month - last_monthbreak_kk_add + kk_add + if not finish: + diff += kk_minimum_tax + final_minus = expenses_so_far - old_needed_income_before_anything + diff + diff += kk_minimum_tax + final_minus = expenses_so_far - old_needed_income_before_anything + diff + self.transfer_lines += [TransferLine(None, acc_assets, -diff, 'â¬')] + self.transfer_lines += [TransferLine(None, acc_assets, final_minus, 'â¬')] + self.transfer_lines += [TransferLine(None, acc_buffer, -final_minus, 'â¬')] + + +# class Booking: +# +# def __init__(self, date_string, description, booking_lines, start_line, process=True): +# self.date_string = date_string +# self.description = description +# self.lines = booking_lines +# self.start_line = start_line +# if process: +# self.validate_booking_lines() +# self.sink = {} +# self.account_changes = self.parse_booking_lines_to_account_changes() +# +# def validate_booking_lines(self): +# prefix = f"booking at line {self.start_line}" +# sums = {} +# empty_values = 0 +# for line in self.lines[1:]: +# if line == '': +# continue +# _, amount, currency = line +# if amount is None: +# if empty_values > 0: +# raise PlomException(f"{prefix} relates more than one empty value of same currency {currency}") +# empty_values += 1 +# continue +# if currency not in sums: +# sums[currency] = 0 +# sums[currency] += amount +# if empty_values == 0: +# for k, v in sums.items(): +# if v != 0: +# raise PlomException(f"{prefix} does not add up to zero / {k} {v}") +# else: +# sinkable = False +# for k, v in sums.items(): +# if v != 0: +# sinkable = True +# if not sinkable: +# raise PlomException(f"{prefix} has empty value that cannot be filled") +# +# def parse_booking_lines_to_account_changes(self): +# account_changes = {} +# debt = {} +# sink_account = None +# for line in self.lines[1:]: +# if line == '': +# continue +# account, amount, currency = line +# if amount is None: +# sink_account = account +# continue +# apply_booking_to_account_balances(account_changes, account, currency, amount) +# if currency not in debt: +# debt[currency] = amount +# else: +# debt[currency] += amount +# if sink_account: +# for currency, amount in debt.items(): +# apply_booking_to_account_balances(account_changes, sink_account, currency, -amount) +# self.sink[currency] = -amount +# return account_changes + + + +class LedgerDB(PlomDB): + + def __init__(self, prefix, ignore_editable_exceptions=False): + self.prefix = prefix + self.bookings = [] + self.comments = [] + self.text_lines = [] + super().__init__(db_path) + self.bookings = parse_lines_to_bookings(self.text_lines, ignore_editable_exceptions) + + def read_db_file(self, f): + self.text_lines += f.readlines() + # self.text_lines += [l.rstrip() for l in f.readlines()] # TODO is this necessary? (parser already removes lines?) + + # def get_lines(self, start, end): + # return self.text_lines[start:end] + + # def write_db(self, text, mode='w'): + # if text[-1] != '\n': + # text += '\n' + # self.write_text_to_db(text) + + # def insert_at_date(self, lines, date): + # start_at = 0 + # if len(self.bookings) > 0: + # if date >= self.bookings[-1].date_string: + # start_at = len(self.text_lines) + # lines = [''] + lines + # else: + # for b in self.bookings: + # if b.date_string == date: + # start_at = b.start_line + # elif b.date_string > date: + # start_at = b.start_line + # break + # lines += [''] # DEBUG is new + # return self.write_lines_in_total_lines_at(self.text_lines, start_at, lines) + + # def update(self, start, end, lines, date): + # remaining_lines = self.text_lines[:start] + self.text_lines[end:] + # n_original_lines = end - start + # start_at = len(remaining_lines) + # for b in self.bookings: + # if b.date_string == date: + # if start_at == len(remaining_lines) or b.start_line == start: + # start_at = b.start_line + # if b.start_line > start: + # start_at -= n_original_lines + # elif b.date_string > date: + # break + # # print("DEBUG update start_at", start_at, "len(remaining_lines)", len(remaining_lines), "len(self.text_lines)", len(self.text_lines), "end", end) + # if start_at != 0 and end != len(self.text_lines) and start_at == len(remaining_lines): + # # Add empty predecessor line if appending. + # lines = [''] + lines + # return self.write_lines_in_total_lines_at(remaining_lines, start_at, lines) + + # def write_lines_in_total_lines_at(self, total_lines, start_at, lines): + # # total_lines = total_lines[:start_at] + lines + [''] + total_lines[start_at:] + # total_lines = total_lines[:start_at] + lines + total_lines[start_at:] + # _, _ = parse_lines(lines) + # text = '\n'.join(total_lines) + # self.write_db(text) + # return start_at + + # def get_nth_for_booking_of_start_line(self, start_line): + # nth = 0 + # for b in self.bookings: + # if b.start_line >= start_line: + # break + # nth += 1 + # return nth + + def insert_booking_at_date(self, booking): + place_at = 0 + if len(self.bookings) > 0: + for i, iterated_booking in enumerate(self.bookings): + if booking.date < iterated_booking.date: + break + elif booking.date == iterated_booking.date: + place_at = i + break 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 + place_at = i + 1 + self.bookings.insert(place_at, booking) + + def add_taxes(self, lines, finish=False): + ret = [] + bookings, _ = parse_lines(lines) + date = bookings[0].date_string + acc_kk_add = 'Reserves:KrankenkassenBeitragsWachstum' + acc_kk_minimum = 'Reserves:Month:KrankenkassenDefaultBeitrag' + acc_kk = 'Expenses:KrankenKasse' + acc_est = 'Reserves:Einkommenssteuer' + acc_assets = 'Assets' + acc_buffer = 'Reserves:NeuAnfangsPuffer:Ausgaben' + last_monthbreak_assets = 0 + last_monthbreak_est = 0 + last_monthbreak_kk_minimum = 0 + last_monthbreak_kk_add = 0 + buffer_expenses = 0 + kk_expenses = 0 + est_expenses = 0 + months_passed = -int(finish) + for b in self.bookings: + if date == b.date_string: + break + acc_keys = b.account_changes.keys() + if acc_buffer in acc_keys: + buffer_expenses -= b.account_changes[acc_buffer]['â¬'] + if acc_kk_add in acc_keys: + kk_expenses += b.account_changes[acc_kk_add]['â¬'] + if acc_kk in acc_keys: + kk_expenses += b.account_changes[acc_kk]['â¬'] + if acc_est in acc_keys: + est_expenses += b.account_changes[acc_est]['â¬'] + if acc_kk_add in acc_keys and acc_kk_minimum in acc_keys: + months_passed += 1 + if finish: + last_monthbreak_kk_add = b.account_changes[acc_kk_add]['â¬'] + last_monthbreak_est = b.account_changes[acc_est]['â¬'] + last_monthbreak_kk_minimum = b.account_changes[acc_kk_minimum]['â¬'] + last_monthbreak_assets = b.account_changes[acc_buffer]['â¬'] + old_needed_income_before_anything = - last_monthbreak_assets - last_monthbreak_kk_add - last_monthbreak_kk_minimum - last_monthbreak_est + if finish: + ret += [f' {acc_est} {-last_monthbreak_est}⬠; for old assumption of needed income: {old_needed_income_before_anything}â¬'] + _, account_sums = bookings_to_account_tree(bookings) + expenses_so_far = -1 * account_sums[acc_assets]['â¬'] + old_needed_income_before_anything + needed_income_before_kk = expenses_so_far + ESt_this_month = 0 + left_over = needed_income_before_kk - ESt_this_month + too_low = 0 + too_high = 2 * needed_income_before_kk + E0 = decimal.Decimal(10908) + E1 = decimal.Decimal(15999) + E2 = decimal.Decimal(62809) + E3 = decimal.Decimal(277825) + while True: + zvE = buffer_expenses - kk_expenses + (12 - months_passed) * needed_income_before_kk + if finish: + zvE += last_monthbreak_assets + last_monthbreak_kk_add + last_monthbreak_kk_minimum + 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: - 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]) + ESt = decimal.Decimal(0.45) * (zvE - decimal.Decimal(277825)) + decimal.Decimal(106713.52) + ESt_this_month = (ESt + last_monthbreak_est - est_expenses) / (12 - months_passed) + left_over = needed_income_before_kk - ESt_this_month + if abs(left_over - expenses_so_far) < 0.001: + break + elif left_over < expenses_so_far: + too_low = needed_income_before_kk + elif left_over > expenses_so_far: + too_high = needed_income_before_kk + needed_income_before_kk = too_low + (too_high - too_low)/2 + ESt_this_month = ESt_this_month.quantize(decimal.Decimal('0.00')) + ret += [f' {acc_est} {ESt_this_month}⬠; expenses so far: {expenses_so_far:.2f}â¬; zvE: {zvE:.2f}â¬; ESt total: {ESt:.2f}â¬; needed before Krankenkasse: {needed_income_before_kk:.2f}â¬'] + kk_minimum_income = 1131.67 + if date < '2023-02-01': + kk_minimum_income = decimal.Decimal(1096.67) + kk_factor = decimal.Decimal(0.189) + kk_minimum_tax = decimal.Decimal(207.27).quantize(decimal.Decimal('0.00')) + elif date < '2023-08-01': + kk_factor = decimal.Decimal(0.191) + kk_minimum_tax = decimal.Decimal(216.15).quantize(decimal.Decimal('0.00')) + else: + kk_factor = decimal.Decimal(0.197) + kk_minimum_tax = decimal.Decimal(222.94).quantize(decimal.Decimal('0.00')) + kk_add_so_far = account_sums[acc_kk_add]['â¬'] if acc_kk_add in account_sums.keys() else 0 + kk_add = needed_income_before_kk / (1 - kk_factor) - needed_income_before_kk - kk_minimum_tax + hit_kk_minimum_income_limit = False + if kk_add_so_far + kk_add < 0: + hit_kk_minimum_income_limit = True + kk_add_uncorrect = kk_add + kk_add = -(kk_add + kk_add_so_far) + kk_add = decimal.Decimal(kk_add).quantize(decimal.Decimal('0.00')) + if finish: + ret += [f' {acc_kk_add} {-last_monthbreak_kk_add}⬠; for old assumption of needed income'] + else: + ret += [f' {acc_kk_minimum} {kk_minimum_tax}⬠; assumed minimum income {kk_minimum_income:.2f}⬠* {kk_factor:.3f}'] + if hit_kk_minimum_income_limit: + ret += [f' {acc_kk_add} {kk_add}⬠; {needed_income_before_kk:.2f}⬠/ (1 - {kk_factor:.3f}) - {needed_income_before_kk:.2f}⬠- {kk_minimum_tax}⬠= {kk_add_uncorrect:.2f}⬠would reduce current {acc_kk_dd} ({kk_add_so_far:.2f}â¬) below 0'] + else: + ret += [f' {acc_kk_add} {kk_add}⬠; {needed_income_before_kk:.2f}⬠/ (1 - {kk_factor:.3f}) - {needed_income_before_kk:.2f}⬠- {kk_minimum_tax}â¬'] + diff = - last_monthbreak_est + ESt_this_month - last_monthbreak_kk_add + kk_add + if not finish: + diff += kk_minimum_tax + final_minus = expenses_so_far - old_needed_income_before_anything + diff + ret += [f' {acc_assets} {-diff} â¬'] + ret += [f' {acc_assets} {final_minus} â¬'] + year_needed = buffer_expenses + final_minus + (12 - months_passed - 1) * final_minus + if finish: + ret += [f' {acc_buffer} {-final_minus} â¬'] + else: + ret += [f' {acc_buffer} {-final_minus} ⬠; assume as to earn in year: {acc_buffer} + {12 - months_passed - 1} * this = {year_needed}'] + return ret + + # def add_mirror(self, lines): + # ret = [] + # bookings, _ = parse_lines(lines) + # booking = bookings[0] + # for line in booking.lines[1:]: + # ret += [f'? {-line[1]} {line[2]}'] + # return ret + + def ledger_as_html(self): + for index, booking in enumerate(self.bookings): + booking.can_up = index > 0 and self.bookings[index - 1].date == booking.date + booking.can_down = index < len(self.bookings) - 1 and self.bookings[index + 1].date == booking.date + return j2env.get_template('ledger.html').render(bookings=self.bookings) + + # def ledger_as_html(self): + # booking_tmpl = jinja2.Template(booking_html) + # single_c_tmpl = jinja2.Template('{{c|e}}
{content}" + + def edit(self, index, sent=None, error_msg=None, edit_mode='table', copy=False): + accounts = set() + if sent or -1 == index: + content = sent if sent else ([] if 'textarea'==edit_mode else None) + else: + content = self.bookings[index] + date_today = str(datetime.now())[:10] + if copy: + content.date = date_today + elif -1 == index and (content is None or [] == content): + content = Booking(date=date_today, validate=False) + if 'textarea' == edit_mode and content: + content = content.for_writing + else: + for booking in self.bookings: + for transfer_line in booking.transfer_lines: + accounts.add(transfer_line.account) + 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)) + + # def add_free(self, start=0, end=0, copy=False): + # tmpl = jinja2.Template(add_form_header + add_free_html + add_form_footer) + # lines = self.get_lines(start, end) + # if copy: + # start = end = 0 + # return tmpl.render(action=self.prefix + '/add_free', start=start, end=end, lines=lines) + + # def add_structured(self, start=0, end=0, copy=False, temp_lines=[], add_empty_line=None): + # tmpl = jinja2.Template(add_form_header + add_structured_html + add_form_footer) + # lines = temp_lines if len(''.join(temp_lines)) > 0 else self.get_lines(start, end) + # bookings, comments = parse_lines(lines, validate_bookings=False) + # if len(bookings) > 1: + # raise PlomException('can only structurally edit single Booking') + # if add_empty_line is not None: + # comments = comments[:add_empty_line+1] + [''] + comments[add_empty_line+1:] + # booking = bookings[0] + # booking.lines = booking.lines[:add_empty_line+1] + [''] + booking.lines[add_empty_line+1:] + # action = self.prefix + '/add_structured' + # datalist_sets = {'descriptions': set(), 'accounts': set(), 'currencies': set()} + # for b in self.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) + # content = '' + # today = str(datetime.now())[:10] + # booking_lines = [] + # if copy: + # start = end = 0 + # desc = head_comment = '' + # if len(bookings) == 0: + # date=today + # else: + # booking = bookings[0] + # desc = booking.description + # date = today if copy else booking.date_string + # head_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] + # booking_lines += [{ + # 'i': i, + # 'acc': account, + # 'amt': amount, + # 'curr': currency if currency else 'â¬', + # 'comment': comments[i], + # 'comm_cols': len(comments[i])}] + # for i in range(len(comments), len(comments) + 8): + # booking_lines += [{'i': i, 'acc': '', 'amt': '', 'curr': 'â¬', 'comment': ''}] + # content += tmpl.render( + # action=action, + # date=date, + # desc=desc, + # head_comment=head_comment, + # booking_lines=booking_lines, + # datalist_sets=datalist_sets, + # start=start, + # end=end) + # return content + + def move_up(self, index): + return self.move(index, -1) + + def move_down(self, index): + return self.move(index, +1) + + def move(self, index, direction): + to_move = self.bookings[index] + swap_index = index + 1*(direction) + to_swap = self.bookings[swap_index] + self.bookings[index] = to_swap + self.bookings[index + 1*(direction)] = to_move + return swap_index + + def write_db(self): lines = [] - def print_subtree(lines, indent, node, subtree, path): - line = f"{indent}{node}" - n_tabs = 5 - (len(line) // 8) - line += n_tabs * "\t" - if "â¬" in account_sums[path + node].keys(): - amount = account_sums[path + node]["â¬"] - line += f"{amount:9.2f} â¬\t" + for i, booking in enumerate(self.bookings): + if i > 0: + lines += [''] + lines += booking.for_writing + self.write_text_to_db('\n'.join(lines) + '\n') + + # def move_up(self, start, end): + # prev_booking = None + # for redir_nth, b in enumerate(self.bookings): + # if b.start_line >= start: + # break + # prev_booking = b + # start_at = prev_booking.start_line + # self.make_move(start, end, start_at) + # return redir_nth - 1 + + # def move_down(self, start, end): + # next_booking = None + # for redir_nth, b in enumerate(self.bookings): + # if b.start_line > start: + # next_booking = b + # break + # # start_at = next_booking.start_line + len(next_booking.lines) - (end - start) + 1 + # # self.make_move(start, end, start_at-1) + # start_at = next_booking.start_line + len(next_booking.lines) - (end - start) + # self.make_move(start, end, start_at) + # return redir_nth + + # def make_move(self, start, end, start_at): + # # FIXME currently broken due to changed self.write_lines_in_total_lines_at, easy fix would be lines += [""] maybe? + # lines = self.get_lines(start, end) + # if start == 0: + # total_lines = self.text_lines[end+1:] + # lines = [''] + lines + # start_at += 1 + # else: + # total_lines = self.text_lines[:start-1] + self.text_lines[end:] # -1 because we reduce the original position's two empty limit lines to one in-between line + # lines += [''] + # self.write_lines_in_total_lines_at(total_lines, start_at, lines) + + # def booking_lines_from_postvars(self, postvars): + # add_empty_line = None + # date = postvars['date'][0] + # description = postvars['description'][0] + # start_comment = postvars['line_0_comment'][0] + # start_line = f'{date} {description}' + # if start_comment.rstrip() != '': + # start_line += f' ; {start_comment}' + # lines = [start_line] + # if 'line_0_add' in postvars.keys(): + # add_empty_line = 0 + # i = j = 1 + # while f'line_{i}_comment' in postvars.keys(): + # if f'line_{i}_delete' in postvars.keys(): + # i += 1 + # continue + # elif f'line_{i}_delete_after' in postvars.keys(): + # break + # elif f'line_{i}_add' in postvars.keys(): + # add_empty_line = j + # 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}' + # if '' == new_main.rstrip() == comment.rstrip(): # don't write empty lines, ignore currency if nothing else set + # continue + # if len(amount.rstrip()) > 0: + # new_main += f' {currency}' + # j += 1 + # new_line = new_main + # if comment.rstrip() != '': + # new_line += f' ; {comment}' + # lines += [new_line] + # if 'add_sink' in postvars.keys(): + # temp_lines = lines.copy() + ['_'] + # try: + # temp_bookings, _ = parse_lines(temp_lines) + # for currency in temp_bookings[0].sink: + # amount = temp_bookings[0].sink[currency] + # # lines += [f'Assets {amount:.2f} {currency}'] + # lines += [f'Assets {amount} {currency}'] + # except PlomException: + # pass + # elif 'add_taxes' in postvars.keys(): + # lines += self.add_taxes(lines, finish=False) + # elif 'add_taxes2' in postvars.keys(): + # lines += self.add_taxes(lines, finish=True) + # elif 'replace' in postvars.keys(): + # for i, line in enumerate(lines): + # lines[i] = line.replace(postvars['replace_from'][0], postvars['replace_to'][0]) + # elif 'add_mirror' in postvars.keys(): + # lines += self.add_mirror(lines) + # return lines, add_empty_line + + + +class LedgerHandler(PlomHandler): + + def app_init(self, handler): + default_path = '/ledger' + handler.add_route('GET', default_path, self.forward_gets) + handler.add_route('POST', default_path, self.forward_posts) + return 'ledger', default_path + + def do_POST(self): + self.try_do(self.forward_posts) + + def forward_posts(self): + prefix = self.apps['ledger'] if hasattr(self, 'apps') else '' + length = int(self.headers['content-length']) + postvars = parse_qs(self.rfile.read(length).decode(), keep_blank_values=1) + db = LedgerDB(prefix, ignore_editable_exceptions=True) + index = 0 + parsed_url = urlparse(self.path) + for string in {'update', 'add', 'check', 'mirror', 'fill_sink', 'textarea', 'table', 'move_up', 'move_down', 'add_taxes', 'replace'}: + if string in postvars.keys(): + submit_button = string + break + if f'{prefix}/ledger' == parsed_url.path and submit_button in {'move_up', 'move_down'}: + mover = getattr(db, submit_button) + index = mover(int(postvars[submit_button][0])) + elif prefix + '/edit' == parsed_url.path: + index = int(postvars['index'][0]) + edit_mode = postvars['edit_mode'][0] + validate = submit_button in {'update', 'add', 'copy', 'check'} + starts_at = '?' if index == -1 else db.bookings[index].starts_at + if 'textarea' == edit_mode: + lines = [LedgerTextLine(line) for line in postvars['booking'][0].rstrip().split('\n')] + booking = Booking(lines, starts_at, validate=validate) else: - line += f"\t\t" - for currency, amount in account_sums[path + node].items(): - if currency != 'â¬' and amount > 0: - line += f"{amount:5.2f} {currency}\t" - lines += [line] - indent += " " - for k, v in sorted(subtree.items()): - print_subtree(lines, indent, k, v, path + node + ":") - for k, v in sorted(account_tree.items()): - print_subtree(lines, "", k, v, "") - content = "\n".join(lines) - return f"
{content}" - - -class MyServer(BaseHTTPRequestHandler): + booking = Booking.from_postvars(postvars, starts_at, validate) + if submit_button in {'update', 'add'}: + if submit_button == 'update': + if 'textarea' == edit_mode and 'delete' == ''.join([l.text_line for l in lines]).strip(): + del db.bookings[index] + # if not creating new Booking, and date unchanged, keep it in place + elif booking.date == db.bookings[index].date: + db.bookings[index] = booking + else: + del db.bookings[index] + db.insert_booking_at_date(booking) + else: + db.insert_booking_at_date(booking) + else: # non-DB-writing calls + error_msg = None + if 'check' == submit_button: + error_msg = 'All looks fine!' + elif submit_button in {'mirror', 'fill_sink', 'add_taxes'}: + getattr(booking, submit_button)() + elif 'replace' == submit_button: + booking.replace(postvars['replace_from'][0], postvars['replace_to'][0]) + elif submit_button in {'textarea', 'table'}: + edit_mode = submit_button + page = db.edit(index, booking, error_msg=error_msg, edit_mode=edit_mode) + self.send_HTML(page) + return + db.write_db() + index = index if index >= 0 else len(db.bookings) - 1 + self.redirect(prefix + f'/ledger#{index}') + + # def forward_posts(self): + # prefix = self.apps['ledger'] if hasattr(self, 'apps') else '' + # parsed_url = urlparse(self.path) + # length = int(self.headers['content-length']) + # postvars = parse_qs(self.rfile.read(length).decode(), keep_blank_values=1) + # start = int(postvars['start'][0]) + # end = int(postvars['end'][0]) + # print("DEBUG start, end", start, end) + # db = LedgerDB(prefix) + # add_empty_line = None + # lines = [] + # # get inputs + # if prefix + '/add_structured' == parsed_url.path and not 'revert' in postvars.keys(): + # lines, add_empty_line = db.booking_lines_from_postvars(postvars) + # elif prefix + '/add_free' == parsed_url.path and not 'revert' in postvars.keys(): + # lines = postvars['booking'][0].splitlines() + # # validate where appropriate + # if ('save' in postvars.keys()) or ('check' in postvars.keys()): + # _, _ = parse_lines(lines) + # # if saving, process where to and where to redirect after + # if 'save' in postvars.keys(): + # last_date = str(datetime.now())[:10] + # if len(db.bookings) > 0: + # last_date = db.bookings[-1].date_string + # target_date = last_date[:] + # first_line_tokens = lines[0].split() if len(lines) > 0 else '' + # first_token = first_line_tokens[0] if len(first_line_tokens) > 0 else '' + # try: + # datetime.strptime(first_token, '%Y-%m-%d') + # target_date = first_token + # except ValueError: + # pass + # if start == end == 0: + # start = db.insert_at_date(lines, target_date) + # nth = db.get_nth_for_booking_of_start_line(start) + # else: + # new_start = db.update(start, end, lines, target_date) + # print("DEBUG save", new_start, start, end, lines) + # nth = db.get_nth_for_booking_of_start_line(new_start) + # if new_start > start: + # nth -= 1 + # self.redirect(prefix + f'/#{nth}') + # # otherwise just re-build editing form + # else: + # if prefix + '/add_structured' == parsed_url.path: + # edit_content = db.add_structured(start, end, temp_lines=lines, add_empty_line=add_empty_line) + # else: + # edit_content = db.add_free(start, end) + # header = jinja2.Template(html_head).render(prefix=prefix) + # self.send_HTML(header + edit_content) def do_GET(self): - self.send_response(200) - self.send_header("Content-type", "text/html") - self.end_headers() - db = Database() - # page = db.ledger_as_html() - page = db.balance_as_html() - self.wfile.write(bytes(page, "utf-8")) - - -db = Database() -if __name__ == "__main__": - webServer = HTTPServer((hostName, serverPort), MyServer) - print(f"Server started http://{hostName}:{serverPort}") - try: - webServer.serve_forever() - except KeyboardInterrupt: - pass - webServer.server_close() - print("Server stopped.") + self.try_do(self.forward_gets) + + def forward_gets(self): + prefix = self.apps['ledger'] if hasattr(self, 'apps') else '' + try: + db = LedgerDB(prefix=prefix) + except EditableException as e: + # We catch the EditableException for further editing, and then + # re-run the DB initiation without it blocking DB creation. + db = LedgerDB(prefix=prefix, ignore_editable_exceptions=True) + page = db.edit(index=e.booking_index, error_msg=f'ERROR: {e}') + self.send_HTML(page) + return + parsed_url = urlparse(self.path) + params = parse_qs(parsed_url.query) + if parsed_url.path == f'{prefix}/balance': + stop = params.get('until_after', [None])[0] + page = db.balance_as_html(stop) + elif parsed_url.path == f'{prefix}/edit': + index = params.get('i', [-1])[0] + copy = params.get('copy', [0])[0] + page = db.edit(int(index), copy=bool(copy)) + else: + page = db.ledger_as_html() + self.send_HTML(page) + + + +if __name__ == "__main__": + run_server(server_port, LedgerHandler)