Alle drei Bank-Parser gegen die realen Fixtures gehaertet: - DKB/VR/HVB: Sammlung endet am Schlusssaldo (break), damit Summen, Rechnungsabschluss und Hinweisseiten nicht mehr in die letzte Buchung rutschen. - Generische, rein strukturelle Noise-Filter fuer wiederkehrende Seitenfuss-/Kopfbloecke und Blattwechsel-Reste (keine Kontodaten hart codiert); Kopf-/Hinweisbloecke werden als Sequenz uebersprungen. - VR: Rekonstruktion harter Zeilenumbrueche im Verwendungszweck unter konservativen Schutzbedingungen (nur purpose, nicht counterparty). - Neue Noise- und expected-Value-Tests; expected_*.json gitignored. - Audit-Werkzeug scripts/parser_audit.py ergaenzt (nur lokale Nutzung). Transaktionszahlen und Saldo-Differenzen unveraendert; max. Laenge des Verwendungszwecks je Bank deutlich unter 300 Zeichen. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
144 lines
6.6 KiB
Python
144 lines
6.6 KiB
Python
"""Parser für HypoVereinsbank (UniCredit)-Kontoauszüge.
|
|
|
|
Realer Fixture-Abgleich (Task 7, Schritt 4): Das Layout weicht strukturell vom
|
|
generischen Platzhalter ab, nicht nur in der Wortwahl:
|
|
- Eine Buchung besteht aus zwei Zeilen: Zeile 1 = Buchungsdatum + Vorgangsart +
|
|
Saldo/Umsatz-Betrag mit direkt angehängtem Vorzeichen (kein Leerzeichen vor
|
|
"+"/"-"); Zeile 2 = Wertstellungsdatum + Buchungsinformation (Text kann
|
|
leer sein, dann folgt nur ein Datum ohne weiteren Text). Fehlt Zeile 2 ganz
|
|
(z. B. Seitenumbruch mitten in der Buchung), gilt das als Parserfehler statt
|
|
eine unvollständige Buchung zu erzeugen.
|
|
- Mehrseitige Auszüge wiederholen Kopfzeilen und Zwischensummen
|
|
("Übertrag: EUR ...", "Saldo per DATUM: EUR ..."), die nicht mit dem
|
|
tatsächlichen Anfangs-/Endsaldo verwechselt werden dürfen und aus dem
|
|
Verwendungszweck herausgefiltert werden.
|
|
- Der Auszugszeitraum steht nur in der wiederholten Kopfzeile als
|
|
"<Start> - <Ende> <Auszug-Nr.> <Seite>".
|
|
"""
|
|
import re
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
import pdfplumber
|
|
|
|
from app.parsers.base import (ParsedStatement, ParsedTransaction, ParserError,
|
|
parse_german_amount, parse_german_date)
|
|
|
|
BANK = "hvb"
|
|
RE_IBAN = re.compile(r"\b([A-Z]{2}\d{2}(?:\s?\d{4}){4,5}(?:\s?\d{1,2})?)\b")
|
|
# Kopfzeile pro Seite: "... <Zeitraum-Start> - <Zeitraum-Ende> <Auszug-Nr> <Seite>"
|
|
RE_PERIOD = re.compile(r"(\d{2}\.\d{2}\.\d{4})\s*-\s*(\d{2}\.\d{2}\.\d{4})\s+\d+\s+\d+\s*$", re.M)
|
|
# "Ihr alter Kontostand per 01.06.2020: EUR 1.234,56+"
|
|
RE_OPENING = re.compile(r"Ihr\s+alter\s+Kontostand\s+per\s+\d{2}\.\d{2}\.\d{4}:\s*EUR\s+([\d.,]+[+-]?)", re.I)
|
|
# "Ihr neuer Kontostand: EUR 2.345,67+" (Zwischensummen heißen "Saldo per ...:" und matchen hier nicht)
|
|
RE_CLOSING = re.compile(r"Ihr\s+neuer\s+Kontostand:\s*EUR\s+([\d.,]+[+-]?)", re.I)
|
|
# Buchungszeile Teil 1: "01.06. LASTSCHRIFT 123,45-" (Buchungsdatum, Vorgang, Betrag+Vorzeichen ohne Leerzeichen)
|
|
RE_TX_HEAD = re.compile(r"^(\d{2}\.\d{2}\.)\s+(.+?)\s+([\d.,]+[+-])$")
|
|
# Buchungszeile Teil 2: "01.06. Muster GmbH" (Wertdatum, Buchungsinformation; Text darf leer sein)
|
|
RE_TX_DETAIL = re.compile(r"^(\d{2}\.\d{2}\.)\s*(.*)$")
|
|
# Zwischensummen/Übertrag zwischen Seiten, dürfen nicht in den Verwendungszweck rutschen
|
|
RE_CARRYOVER = re.compile(r"^(?:Übertrag|Saldo\s+per)\b", re.I)
|
|
# Wiederkehrende Seiten-Kopf-/Fußzeilen (Titel, Spaltenköpfe, Fußnoten), ebenfalls kein Buchungstext
|
|
RE_PAGE_NOISE = re.compile(
|
|
r"^(?:K\s?O\s?N\s?T\s?O\s?A\s?U\s?S\s?Z\s?U\s?G"
|
|
r"|Kontonummer\s+Bankleitzahl"
|
|
r"|IBAN\s+BIC"
|
|
r"|Kontoinhaber\s+Ansprechpartner"
|
|
r"|Buchung/Wert\s+Buchungsinformation"
|
|
r"|Bitte\s+R(?:ü|ue)ckseite\s+beachten"
|
|
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_HEADER_EXTRA.search(line) or RE_PERIOD.search(line))
|
|
|
|
|
|
def parse(path: Path) -> ParsedStatement:
|
|
with pdfplumber.open(path) as pdf:
|
|
text = "\n".join((page.extract_text() or "") for page in pdf.pages)
|
|
period = RE_PERIOD.search(text)
|
|
if not period:
|
|
raise ParserError("HVB: Abrechnungszeitraum nicht gefunden")
|
|
period_start = parse_german_date(period.group(1))
|
|
period_end = parse_german_date(period.group(2))
|
|
opening = RE_OPENING.search(text)
|
|
closing = RE_CLOSING.search(text)
|
|
if not opening or not closing:
|
|
raise ParserError("HVB: Anfangs-/Endsaldo nicht gefunden")
|
|
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:
|
|
if pending["value"] is None:
|
|
raise ParserError("HVB: Buchungszeile ohne Detailzeile")
|
|
txs.append(_finish(pending))
|
|
pending = {
|
|
"booking": parse_german_date(m.group(1), period_end.year),
|
|
"value": None,
|
|
"head": m.group(2).strip(),
|
|
"amount": parse_german_amount(m.group(3)),
|
|
"extra": [],
|
|
}
|
|
continue
|
|
if pending and pending["value"] is None:
|
|
d = RE_TX_DETAIL.match(line)
|
|
if d:
|
|
pending["value"] = parse_german_date(d.group(1), period_end.year)
|
|
if d.group(2).strip():
|
|
pending["extra"].append(d.group(2).strip())
|
|
continue
|
|
if pending and not _is_noise(line):
|
|
pending["extra"].append(line)
|
|
if pending:
|
|
if pending["value"] is None:
|
|
raise ParserError("HVB: Buchungszeile ohne Detailzeile")
|
|
txs.append(_finish(pending))
|
|
return ParsedStatement(bank=BANK,
|
|
iban=iban_m.group(1).replace(" ", "") if iban_m else None,
|
|
period_start=period_start, period_end=period_end,
|
|
opening_balance=parse_german_amount(opening.group(1)),
|
|
closing_balance=parse_german_amount(closing.group(1)),
|
|
transactions=txs)
|
|
|
|
|
|
def _finish(p: dict) -> ParsedTransaction:
|
|
counterparty = p["extra"][0] if p["extra"] else ""
|
|
purpose = " ".join([p["head"], *p["extra"][1:]]).strip()
|
|
return ParsedTransaction(booking_date=p["booking"], value_date=p["value"],
|
|
amount=p["amount"], purpose=purpose,
|
|
counterparty=counterparty)
|