consors-report.py

   1 #!/usr/bin/python3
   2
   3 import argparse
   4 import collections
   5 import csv
   6 import itertools
   7 import re
   8 from collections import namedtuple
   9 from pathlib import Path
  10 from typing import Iterable
  11
  12 Entry = namedtuple('Entry', ['category', 'date', 'valuta', 'who', 'iban', 'bic', 'type', 'desc', 'value'])
  13
  14 CATEGORIES = {
  15     'Gehalt/Steuern': re.compile('Gehalt/Rente|RED HAT|Finanzamt|FK Guenzburg', re.I),
  16     'Wohnung': re.compile('Telekom|hands4home|Goetzfried|Green Planet Energy|Beitragsservice.*Rundfunk', re.I),
  17     'Medizin': re.compile(r'Apotheke|MVZ|\bmed\b|ZAB Abrechnung|Caika|HNOeins|PVS|Dr\..*Sellier|'
  18                           r'BFS Health|Streifeneder|Labor|Physio|(Drescher.*Lung)|Osteopath|'
  19                           '(Dr.*Borchers)|(Debeka.*Überweisung)|(DKV.*Überweisung)|Beihilfe|Klinik|'
  20                           'Niklas Hermann', re.I),
  21     'Versicherung': re.compile('((debeka|DKV|Hallesche|Versicherung|Alte Leipziger|ConceptIF|'
  22                                'Baloise).*Lastschrift)|Hallesche.*Bonu', re.I),
  23     'Transport': re.compile('DB Vertrieb|DB Fernverkehr|Deutsche Bahn|Nextbike|Carsharing|Radstation', re.I),
  24     'Lebensmittel': re.compile('BIOS|Bäcker|Baecker|Ruta|Rewe', re.I),
  25     'Eigentumswohnungen': re.compile('Rechnung Darl.-Leistung|Semmelweis|BHP Buerkner', re.I),
  26     'Hobby/Sport': re.compile('Holstein|Mrs. Sporty|Kieser|DJK|Dimaso', re.I),
  27     'Sparen': re.compile('Bausparkasse Mainz AG|FIL Fondsbank|MIG (Fonds|GmbH)|Netfonds AG|Festgeld', re.I),
  28     'Spenden': re.compile('Spende|Signal Foundation|nebenan.de|(paypal.*Sabine.Hossenf)|'
  29                           'Patreon|Umwelthilfe|Foerderer|Malteser|(Aktion Tier.*Mensch)|'
  30                           'campact|Amnesty', re.I),
  31 }
  32
  33
  34 def parse_args() -> argparse.Namespace:
  35     parser = argparse.ArgumentParser(description='Read Consorsbank CSV and generate report')
  36     parser.add_argument('-d', '--date', metavar='REGEX', default='.', help='date filter regex')
  37     parser.add_argument('-s', '--summary', action='store_true', help='only show category sums')
  38     parser.add_argument('csv', type=Path, nargs='+', help='CSV files to parse')
  39     return parser.parse_args()
  40
  41
  42 def get_category(item: str) -> str:
  43     for category, pattern in CATEGORIES.items():
  44         if pattern.search(item):
  45             return category
  46     return 'Sonstiges'
  47
  48
  49 def parse_entry(raw_fields: Iterable[str]) -> Entry:
  50     fields = [f.strip() for f in raw_fields]
  51     # format change in May 2024, adds a 9th field "Währung"; ignore
  52     fields = fields[:8]
  53     # last field is the value, parse as float
  54     value = float(fields.pop().replace('.', '').replace(',', '.'))
  55     # match on who, IBAN, type, or desc
  56     category = get_category(fields[2] + fields[3] + fields[5] + fields[6])
  57     return Entry._make([category, *fields, value])
  58
  59
  60 def parse_csv(path: Path, date_filter: str) -> Iterable[Entry]:
  61     filter_re = re.compile(date_filter, re.I)
  62
  63     def filter_entry(entry: Entry):
  64         # ignore from/to Tagesgeldkonto
  65         if entry.iban == 'DE13760300800853589101':
  66             return None
  67         return filter_re.search(entry.date)
  68
  69     with path.open() as f:
  70         reader = csv.reader(f, delimiter=';')
  71         next(reader)  # skip header
  72         # first line is the column headers, chop it off
  73         entries = map(parse_entry, reader)
  74         # do the actual iteration here, as the files get closed afterwards
  75         return list(filter(filter_entry, entries))
  76
  77
  78 def main():
  79     args = parse_args()
  80
  81     entries = itertools.chain.from_iterable(map(lambda p: parse_csv(p, args.date), args.csv))
  82     by_category = collections.defaultdict(lambda: [])
  83     category_sum = collections.defaultdict(lambda: 0.0)
  84     balance = 0.0
  85     for e in entries:
  86         by_category[e.category].append(e)
  87         category_sum[e.category] += e.value
  88         balance += e.value
  89
  90     if args.summary:
  91         for cat in sorted(by_category):
  92             print(f'| {category_sum[cat]:>8.2f} | {cat:20} |')
  93         print('|----------|----------------------|')
  94         print(f'| {balance:>8.2f} | {"Saldo":20} |')
  95
  96     else:
  97         for cat in sorted(by_category):
  98             print(f'## {cat}: {category_sum[cat]:.2f}\n')
  99             for entry in by_category[cat]:
 100                 print(f'| {entry.date:7} | {entry.value:>7.2f} | _{entry.type}_ - {entry.who} - {entry.desc} |')
 101             print()
 102
 103         print(f'## Saldo: {balance:.2f}')
 104
 105
 106 if __name__ == '__main__':
 107     main()