diff --git a/finance/.gitignore b/finance/.gitignore index e3d2a42..9e5c8d5 100644 --- a/finance/.gitignore +++ b/finance/.gitignore @@ -2,4 +2,5 @@ __pycache__/ *.sqlite tests/fixtures/*.pdf +tests/fixtures/expected_*.json .venv/ diff --git a/finance/app/parsers/dkb.py b/finance/app/parsers/dkb.py index c3fa905..4771a61 100644 --- a/finance/app/parsers/dkb.py +++ b/finance/app/parsers/dkb.py @@ -28,6 +28,25 @@ RE_OPENING = re.compile(r"Kontostand\s+am\s+(\d{2}\.\d{2}\.\d{4}),\s*Auszug\s+Nr RE_CLOSING = re.compile(r"Kontostand\s+am\s+(\d{2}\.\d{2}\.\d{4})\s+um\s+\d{2}:\d{2}\s*Uhr\s+(-?[\d.,]+)", re.I) # Buchungszeile: "01.06.2020Überweisung -123,45" (Datum direkt am Text, kein Leerzeichen) RE_TX_HEAD = re.compile(r"^(\d{2}\.\d{2}\.\d{4})(\S.*?)\s+(-?[\d.,]+)$") +# Wiederkehrender Seitenfuß (Bank-Impressum) und Folgeseiten-Kopf; rein +# strukturelle Muster (keine Kontodaten), damit dieser Block nicht als +# Verwendungszweck der jeweils letzten Buchung einer Seite eingesammelt wird. +RE_NOISE = re.compile( + r"(?:Kreditbank AG" # Impressum-Zeile 1 + r"|Taubenstraße" # Impressum-Zeile 2 (Anschrift) + r"|^\d{5}\s+Berlin\b" # Impressum-Zeile 3 (PLZ Ort) + r"|Handelsregister" # Impressum-Zeile 4 + r"|^Ein Unternehmen" # Impressum-Zeile 5 + r"|Landesbank" # Impressum-Zeile 6 + r"|^Kontoauszug\s+\d+/\d+\s+Seite" # Folgeseiten-Kopf + r"|^Girokonto\s+\d" # Folgeseiten-Kopf (Kontozeile) + r"|^Datum\s+Erläuterung" # Spaltenüberschrift + r")" +) + + +def _is_noise(line: str) -> bool: + return bool(RE_NOISE.search(line)) def parse(path: Path) -> ParsedStatement: @@ -46,6 +65,10 @@ def parse(path: Path) -> ParsedStatement: line = line.strip() if not line: continue + # Schlusssaldo erreicht: laufende Buchung abschließen und Sammlung beenden + # (danach folgen nur noch Summen, Rechnungsabschluss und Hinweise). + if RE_CLOSING.search(line): + break m = RE_TX_HEAD.match(line) if m: if pending: @@ -59,7 +82,7 @@ def parse(path: Path) -> ParsedStatement: "extra": [], } continue - if pending and not RE_OPENING.search(line) and not RE_CLOSING.search(line): + if pending and not RE_OPENING.search(line) and not _is_noise(line): pending["extra"].append(line) if pending: txs.append(_finish(pending)) diff --git a/finance/app/parsers/hvb.py b/finance/app/parsers/hvb.py index 3ffe4ce..e8a3f78 100644 --- a/finance/app/parsers/hvb.py +++ b/finance/app/parsers/hvb.py @@ -49,11 +49,24 @@ RE_PAGE_NOISE = re.compile( r"|Seite\s+\d+\s+von\s+\d+)", re.I, ) +# Weitere strukturelle Kopf-/Fußzeilen am Seitenwechsel (keine Kontodaten hart +# codiert): Formularkennung, Ansprechpartner-Zeile, zusammengesetzte IBAN+BIC-Zeile. +RE_HEADER_EXTRA = re.compile( + r"(?:Smart Banking Team" # Ansprechpartner-/Kopfzeile + r"|^\d{4}\.\d{2}/[A-Z]\d" # Formularkennung (z. B. 5202.70/C0..) + r"|^DE\d{2}\s\d{8}\s\d{10}\s[A-Z]{8}" # IBAN + BIC in einer Zeile + r")" +) +# Ganzseitiger Hinweis-/Rückseitenblock zwischen den Buchungsseiten: als +# Sequenz vom Startsatz bis zur nächsten "Seite N von M"-Zeile überspringen. +RE_HINT_START = re.compile(r"^Wir bitten Sie") +RE_SEITE = re.compile(r"^Seite\s+\d+\s+von\s+\d+") def _is_noise(line: str) -> bool: return bool(RE_OPENING.search(line) or RE_CLOSING.search(line) - or RE_CARRYOVER.match(line) or RE_PAGE_NOISE.match(line)) + or RE_CARRYOVER.match(line) or RE_PAGE_NOISE.match(line) + or RE_HEADER_EXTRA.search(line) or RE_PERIOD.search(line)) def parse(path: Path) -> ParsedStatement: @@ -71,10 +84,22 @@ def parse(path: Path) -> ParsedStatement: iban_m = RE_IBAN.search(text) txs: list[ParsedTransaction] = [] pending: dict | None = None + in_hint = False for line in text.splitlines(): line = line.strip() if not line: continue + # Ganzseitigen Hinweisblock als Sequenz überspringen. + if in_hint: + if RE_SEITE.match(line): + in_hint = False + continue + if RE_HINT_START.match(line): + in_hint = True + continue + # Schlusssaldo erreicht: laufende Buchung abschließen und Sammlung beenden. + if RE_CLOSING.search(line): + break m = RE_TX_HEAD.match(line) if m: if pending: diff --git a/finance/app/parsers/vr.py b/finance/app/parsers/vr.py index f97a2d2..33f3aa0 100644 --- a/finance/app/parsers/vr.py +++ b/finance/app/parsers/vr.py @@ -27,6 +27,50 @@ RE_OPENING = re.compile(r"alter\s+Kontostand\s+(?:vom\s+(\d{2}\.\d{2}\.\d{4})\s+ RE_CLOSING = re.compile(r"neuer\s+Kontostand\s+(?:vom\s+(\d{2}\.\d{2}\.\d{4})\s+)?([\d.,]+\s*[SH-]?)", re.I) # Buchungszeile: "01.06. 01.06. Muster GmbH 1.234,56 S" RE_TX = re.compile(r"^(\d{2}\.\d{2}\.)\s+(\d{2}\.\d{2}\.)\s+(.*?)\s+([\d.,]+\s*[SH-])$") +# Folgeseiten-Kopfblock (Bankanschrift bis Spaltenüberschrift) als Sequenz +# überspringen: fängt auch die Kontoinhaber-Kopfzeile ab, ohne gleichnamige +# Gegenparteien in Buchungen zu zerstören. +RE_HEADER_START = re.compile(r"^Beraterpark am Olgaplatz") +RE_HEADER_END = re.compile(r"^Bu-Tag\s+Wert\s+Vorgang") +# Blattwechsel-/Formular-Reste, die sonst in den Verwendungszweck rutschen. +RE_NOISE = re.compile( + r"(?:Übertrag" # "Übertrag auf/von Blatt N" + r"|^\d{3,4}$" # Formularcodes (z. B. 0007, 000) + r"|^K\d{6}" # Formularcode (K + Ziffernblock) + r"|Bitte beachten Sie die Hinweise" # Fußzeilen-Hinweis + r"|^[─—–\-]{3,}$" # Trennlinien + r")" +) + + +def _is_noise(line: str) -> bool: + return bool(RE_NOISE.search(line)) + + +def _join_purpose(parts: list[str]) -> str: + """Verwendungszweck zusammensetzen und harte Zeilenumbrüche der VR-PDF + (~54 Zeichen, teils mitten im Wort) rekonstruieren. + + Ohne Leerzeichen wird nur dann verbunden, wenn die Vorzeile mindestens 53 + Zeichen lang ist UND (Vorzeile endet auf Ziffer und Folgezeile beginnt mit + Ziffer) ODER (beide an der Bruchstelle Kleinbuchstaben) – sonst mit + Leerzeichen. So bleibt z. B. "... 75" + "Euro" mit Leerzeichen getrennt. + """ + if not parts: + return "" + out = parts[0] + for prev, cur in zip(parts, parts[1:]): + glue = "" + if len(prev) >= 53 and prev and cur: + if (prev[-1].isdigit() and cur[0].isdigit()) or \ + (prev[-1].islower() and cur[0].islower()): + glue = "" + else: + glue = " " + else: + glue = " " + out += glue + cur + return out.strip() def parse(path: Path) -> ParsedStatement: @@ -48,8 +92,23 @@ def parse(path: Path) -> ParsedStatement: iban_m = RE_IBAN.search(text) txs: list[ParsedTransaction] = [] pending: dict | None = None + in_header = False for line in text.splitlines(): - m = RE_TX.match(line.strip()) + line = line.strip() + if not line: + continue + # Folgeseiten-Kopfblock komplett überspringen (inkl. Kontoinhaber-Zeile). + if in_header: + if RE_HEADER_END.match(line): + in_header = False + continue + if RE_HEADER_START.match(line): + in_header = True + continue + # Schlusssaldo erreicht: laufende Buchung abschließen und Sammlung beenden. + if RE_CLOSING.search(line): + break + m = RE_TX.match(line) if m: if pending: txs.append(_finish(pending)) @@ -60,8 +119,8 @@ def parse(path: Path) -> ParsedStatement: "amount": parse_german_amount(m.group(4)), "extra": [], } - elif pending and line.strip() and not RE_CLOSING.search(line): - pending["extra"].append(line.strip()) + elif pending and not _is_noise(line): + pending["extra"].append(line) if pending: txs.append(_finish(pending)) return ParsedStatement(bank=BANK, @@ -74,7 +133,7 @@ def parse(path: Path) -> ParsedStatement: def _finish(p: dict) -> ParsedTransaction: counterparty = p["extra"][0] if p["extra"] else "" - purpose = " ".join([p["head"], *p["extra"][1:]]).strip() + purpose = _join_purpose([p["head"], *p["extra"][1:]]) return ParsedTransaction(booking_date=p["booking"], value_date=p["value"], amount=p["amount"], purpose=purpose, counterparty=counterparty) diff --git a/finance/scripts/parser_audit.py b/finance/scripts/parser_audit.py new file mode 100644 index 0000000..754fe0e --- /dev/null +++ b/finance/scripts/parser_audit.py @@ -0,0 +1,29 @@ +"""Parser-Audit: geparste Transaktionen + Rohtext einer Fixture anzeigen. + +NUR lokal verwenden - Ausgabe enthaelt echte Kontodaten und darf nicht +in Commits, Reports oder Tickets uebernommen werden. +Aufruf: python scripts/parser_audit.py tests/fixtures/dkb_beispiel.pdf +""" +import sys +from pathlib import Path + +import pdfplumber + +from app.parsers.registry import parse_pdf +from app.parsers.validate import balance_difference + +path = Path(sys.argv[1]) +stmt = parse_pdf(path) +print(f"bank={stmt.bank} iban={stmt.iban} " + f"period={stmt.period_start}..{stmt.period_end}") +print(f"opening={stmt.opening_balance} closing={stmt.closing_balance} " + f"diff={balance_difference(stmt)} n_tx={len(stmt.transactions)}") +print("-" * 100) +for i, t in enumerate(stmt.transactions): + print(f"#{i:3d} | {t.booking_date} | {t.value_date} | {t.amount:>12} " + f"| {t.counterparty[:40]:40} | {t.purpose[:60]}") +print("=" * 100) +with pdfplumber.open(path) as pdf: + for n, page in enumerate(pdf.pages, 1): + print(f"--- Seite {n} ---") + print(page.extract_text() or "(kein Text)") diff --git a/finance/tests/test_bank_parsers.py b/finance/tests/test_bank_parsers.py index 90e4a1a..4b5e52a 100644 --- a/finance/tests/test_bank_parsers.py +++ b/finance/tests/test_bank_parsers.py @@ -1,3 +1,5 @@ +import json +import re from decimal import Decimal from pathlib import Path @@ -10,6 +12,9 @@ FIXTURES = Path(__file__).parent / "fixtures" CASES = [("vr_beispiel.pdf", "vr"), ("hvb_beispiel.pdf", "hvb"), ("dkb_beispiel.pdf", "dkb")] +NOISE = re.compile( + r"(Kontostand|Übertrag|alter Saldo|neuer Saldo|Seite \d|Blatt \d)", re.I) + @pytest.mark.parametrize("filename,bank", CASES) def test_parse_fixture(filename, bank): @@ -22,3 +27,32 @@ def test_parse_fixture(filename, bank): assert stmt.closing_balance is not None assert len(stmt.transactions) > 0 assert balance_difference(stmt) == Decimal("0") + + +@pytest.mark.parametrize("filename,bank", CASES) +def test_no_noise_in_parsed_fields(filename, bank): + path = FIXTURES / filename + if not path.exists(): + pytest.skip(f"Fixture {filename} fehlt") + stmt = parse_pdf(path) + for t in stmt.transactions: + assert not NOISE.search(t.purpose), f"Noise im Verwendungszweck: #{stmt.transactions.index(t)}" + assert not NOISE.search(t.counterparty), f"Noise im Empfänger: #{stmt.transactions.index(t)}" + assert stmt.period_start <= t.booking_date <= stmt.period_end + + +@pytest.mark.parametrize("filename,bank", CASES) +def test_expected_values(filename, bank): + path = FIXTURES / filename + exp_path = FIXTURES / f"expected_{bank}.json" + if not path.exists() or not exp_path.exists(): + pytest.skip("Fixture oder expected-Datei fehlt") + stmt = parse_pdf(path) + exp = json.loads(exp_path.read_text()) + assert len(stmt.transactions) == exp["transaction_count"] + for c in exp["spot_checks"]: + t = stmt.transactions[c["index"]] + assert str(t.booking_date) == c["booking_date"] + assert str(t.amount) == c["amount"] + assert c["counterparty_contains"].lower() in t.counterparty.lower() + assert c["purpose_contains"].lower() in t.purpose.lower()