119 lines
5.5 KiB
Python
119 lines
5.5 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,
|
|
)
|
|
|
|
|
|
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)
|