+#!/usr/bin/python3
+
+import argparse
+import collections
+import csv
+import itertools
+import re
+from collections import namedtuple
+from pathlib import Path
+from typing import Iterable
+
+Entry = namedtuple('Entry', ['category', 'date', 'valuta', 'who', 'iban', 'bic', 'type', 'desc', 'value'])
+
+CATEGORIES = {
+ 'Gehalt/Steuern': re.compile('Gehalt/Rente|RED HAT|Finanzamt|FK Guenzburg', re.I),
+ 'Wohnung': re.compile('Telekom|hands4home|Goetzfried|Green Planet Energy|Beitragsservice.*Rundfunk', re.I),
+ 'Medizin': re.compile(r'Apotheke|MVZ|\bmed\b|ZAB Abrechnung|Caika|HNOeins|PVS|Dr\..*Sellier|'
+ r'BFS Health|Streifeneder|Labor|Physio|(Drescher.*Lung)|Osteopath|'
+ '(Dr.*Borchers)|(Debeka.*Überweisung)|(DKV.*Überweisung)|Beihilfe|Klinik|'
+ 'Niklas Hermann|Alena Hauser|Strizrep|Aerzte|Hallesche.*Ausz|Physico|'
+ 'Orthop|Pathologie', re.I),
+ 'Versicherung': re.compile('((debeka|DKV|Hallesche|Versicherung|Alte Leipziger|ConceptIF|'
+ 'Baloise).*Lastschrift)|Hallesche.*Bonu', re.I),
+ 'Transport': re.compile(r'DB Vertr\s*ieb|DB Fernverkehr|Deutsche Bahn|Nextbike|Carsharing|Radstation', re.I),
+ 'Lebensmittel': re.compile('BIOS|Bäcker|Baecker|Baker|Schubert Bio|Ruta|Rewe|Edeka', re.I),
+ 'Eigentumswohnungen': re.compile('Rechnung Darl.-Leistung|Semmelweis|BHP Buerkner|Dr. Schmitz', re.I),
+ 'Hobby/Sport': re.compile('Holstein|Mrs. Sporty|Kieser|DJK|Dimaso', re.I),
+ 'Sparen': re.compile('Bausparkasse Mainz AG|FIL Fondsbank|MIG (Fonds|GmbH)|Netfonds AG|Festgeld|'
+ 'MORGAN STANLEY', re.I),
+ 'Spenden': re.compile('Spende|Signal Foundation|nebenan.de|(paypal.*Sabine.Hossenf)|'
+ 'Patreon|Umwelthilfe|Foerderer|Malteser|(Aktion Tier.*Mensch)|'
+ 'campact|Amnesty|BN Landesverband', re.I),
+}
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser(description='Read Consorsbank CSV and generate report')
+ parser.add_argument('-d', '--date', metavar='REGEX', default='.', help='date filter regex')
+ parser.add_argument('-s', '--summary', action='store_true', help='only show category sums')
+ parser.add_argument('csv', type=Path, nargs='+', help='CSV files to parse')
+ return parser.parse_args()
+
+
+def get_category(item: str) -> str:
+ for category, pattern in CATEGORIES.items():
+ if pattern.search(item):
+ return category
+ return 'Sonstiges'
+
+
+def parse_entry(raw_fields: Iterable[str]) -> Entry:
+ fields = [f.strip() for f in raw_fields]
+ # format change in May 2024, adds a 9th field "Währung"; ignore
+ fields = fields[:8]
+ # last field is the value, parse as float
+ value = float(fields.pop().replace('.', '').replace(',', '.'))
+ # match on who, IBAN, type, or desc
+ category = get_category(fields[2] + fields[3] + fields[5] + fields[6])
+ return Entry._make([category, *fields, value])
+
+
+def parse_csv(path: Path, date_filter: str) -> Iterable[Entry]:
+ filter_re = re.compile(date_filter, re.I)
+
+ def filter_entry(entry: Entry):
+ # ignore from/to Tagesgeldkonto
+ if entry.iban == 'DE13760300800853589101':
+ return None
+ return filter_re.search(entry.date)
+
+ with path.open() as f:
+ reader = csv.reader(f, delimiter=';')
+ next(reader) # skip header
+ # first line is the column headers, chop it off
+ entries = map(parse_entry, reader)
+ # do the actual iteration here, as the files get closed afterwards
+ return list(filter(filter_entry, entries))
+
+
+def main():
+ args = parse_args()
+
+ entries = itertools.chain.from_iterable(map(lambda p: parse_csv(p, args.date), args.csv))
+ by_category = collections.defaultdict(lambda: [])
+ category_sum = collections.defaultdict(lambda: 0.0)
+ balance = 0.0
+ for e in entries:
+ by_category[e.category].append(e)
+ category_sum[e.category] += e.value
+ balance += e.value
+
+ if args.summary:
+ for cat in sorted(by_category):
+ print(f'| {category_sum[cat]:>8.2f} | {cat:20} |')
+ print('|----------|----------------------|')
+ print(f'| {balance:>8.2f} | {"Saldo":20} |')
+
+ else:
+ for cat in sorted(by_category):
+ print(f'## {cat}: {category_sum[cat]:.2f}\n')
+ for entry in by_category[cat]:
+ print(f'| {entry.date:7} | {entry.value:>7.2f} | _{entry.type}_ - {entry.who} - {entry.desc} |')
+ print()
+
+ print(f'## Saldo: {balance:.2f}')
+
+
+if __name__ == '__main__':
+ main()
projects / bin.git / commitdiff