feat: PDF-Parser VR/HVB/DKB mit Registry und Fixture-Tests
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
118
finance/app/parsers/hvb.py
Normal file
118
finance/app/parsers/hvb.py
Normal file
@@ -0,0 +1,118 @@
|
||||
"""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,
|
||||
)
|
||||
|
||||
|
||||
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))
|
||||
|
||||
|
||||
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
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
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)
|
||||
Reference in New Issue
Block a user