185 lines
7.4 KiB
Python
185 lines
7.4 KiB
Python
"""Parser für DKB (Deutsche Kreditbank)-Kontoauszüge.
|
||
|
||
Realer Fixture-Abgleich (Task 7, Schritt 4): Der eigentliche Auszugszeitraum
|
||
steht nicht als "vom X bis Y"-Satz im Text (dieser Wortlaut kommt nur in einem
|
||
separaten quartalsweisen Zinsabschluss-Absatz mit abweichendem Zeitraum vor);
|
||
er wird daher aus den Datumsangaben der beiden Kontostand-Zeilen abgeleitet.
|
||
Buchungszeilen haben nur eine Datumsspalte, die ohne Leerzeichen direkt am
|
||
Buchungstext klebt, und der Betrag trägt ein optionales führendes Minus statt
|
||
eines nachgestellten S/H-Kürzels. Der Anfangssaldo trägt den Zusatz
|
||
", Auszug Nr. N", der Endsaldo den Zusatz "um HH:MM Uhr" – beides
|
||
unterscheidet die echten Saldo-Zeilen von einer abweichend formulierten
|
||
Zwischensumme in einem quartalsweisen Rechnungsabschluss-Anhang.
|
||
"""
|
||
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 = "dkb"
|
||
RE_IBAN = re.compile(r"\b([A-Z]{2}\d{2}(?:\s?\d{4}){4,5}(?:\s?\d{1,2})?)\b")
|
||
# "Kontostand am 01.06.2020, Auszug Nr. 1 1.234,56"
|
||
RE_OPENING = re.compile(r"Kontostand\s+am\s+(\d{2}\.\d{2}\.\d{4}),\s*Auszug\s+Nr\.\s*\d+\s+(-?[\d.,]+)", re.I)
|
||
# "Kontostand am 30.06.2020 um 12:00 Uhr -2.345,67"
|
||
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))
|
||
|
||
|
||
# --- Empfänger/Zweck-Split-Heuristik (D-T5-1/D-T5-2, siehe
|
||
# .superpowers/sdd/parser-vs-csv-audit.md) --------------------------------
|
||
# extras[0] enthält "<Empfänger><Leerzeichen><Beginn Verwendungszweck>" ohne
|
||
# Trenner. Die folgenden Regeln bestimmen, wo der Empfänger endet.
|
||
_LEGAL_FORM_TOKENS = {
|
||
"ag", "gmbh", "ev", "eg", "kg", "bv", "sca", "sarl", "se",
|
||
"ltd", "inc", "ek", "ohg", "gbr", "stiftung", "bank",
|
||
}
|
||
_CONNECTOR_TOKENS = {"et", "cie"}
|
||
_REF_KEYWORDS = {"rechnung", "vertrag", "kundennummer", "kassenzeichen", "rg", "nr"}
|
||
_REF_KEYWORD_PREFIXES = ("kd.-nr", "kdnr", "beitragsnr")
|
||
_RE_TOKEN_DIGIT = re.compile(r"\d")
|
||
_MAX_LEGAL_FORM_SCAN = 7
|
||
|
||
|
||
def _normalize_token(tok: str) -> str:
|
||
return tok.strip(",;").lower().replace(".", "")
|
||
|
||
|
||
def _is_legal_form(tok: str) -> bool:
|
||
return _normalize_token(tok) in _LEGAL_FORM_TOKENS
|
||
|
||
|
||
def _is_connector(tok: str) -> bool:
|
||
return _normalize_token(tok) in _CONNECTOR_TOKENS
|
||
|
||
|
||
def _is_reference_keyword(tok: str) -> bool:
|
||
norm = _normalize_token(tok)
|
||
return norm in _REF_KEYWORDS or norm.startswith(_REF_KEYWORD_PREFIXES)
|
||
|
||
|
||
def _split_counterparty(line: str) -> tuple[str, str]:
|
||
"""Trennt Empfänger von Zweck-Beginn in der ersten Fortsetzungszeile
|
||
(extras[0]) einer DKB-Buchung. Gibt (counterparty, purpose_prefix)
|
||
zurück; purpose_prefix ist der abgeschnittene Rest (kann leer sein).
|
||
|
||
Reihenfolge (D-T5-1):
|
||
1. Rechtsform-Grenze (erste ~7 Tokens; "et Cie"-Ketten bis zum letzten
|
||
Rechtsform-Token in Folge).
|
||
2. Referenz-Grenze (erstes Token mit Ziffer oder Referenz-Schlüsselwort,
|
||
frühestens ab Token 1 -> counterparty nie leer).
|
||
3. Kein Schnitt (reine Wort-Folge, semantisch nicht trennbar).
|
||
|
||
D-T5-2: "siehe Anlage"-Zeilen -> counterparty leer, ganze Zeile in den
|
||
Zweck.
|
||
"""
|
||
line = line.strip()
|
||
if not line:
|
||
return "", ""
|
||
if line.lower().startswith("siehe anlage"):
|
||
return "", line
|
||
tokens = line.split()
|
||
n = len(tokens)
|
||
|
||
for i in range(min(n, _MAX_LEGAL_FORM_SCAN)):
|
||
if not _is_legal_form(tokens[i]):
|
||
continue
|
||
last = i
|
||
j = i + 1
|
||
while j < n and (_is_connector(tokens[j]) or _is_legal_form(tokens[j])):
|
||
if _is_legal_form(tokens[j]):
|
||
last = j
|
||
j += 1
|
||
cut = last + 1
|
||
return " ".join(tokens[:cut]), " ".join(tokens[cut:])
|
||
|
||
for idx in range(1, n):
|
||
if _RE_TOKEN_DIGIT.search(tokens[idx]) or _is_reference_keyword(tokens[idx]):
|
||
return " ".join(tokens[:idx]), " ".join(tokens[idx:])
|
||
|
||
return line, ""
|
||
|
||
|
||
def parse(path: Path) -> ParsedStatement:
|
||
with pdfplumber.open(path) as pdf:
|
||
text = "\n".join((page.extract_text() or "") for page in pdf.pages)
|
||
opening = RE_OPENING.search(text)
|
||
closing = RE_CLOSING.search(text)
|
||
if not opening or not closing:
|
||
raise ParserError("DKB: Anfangs-/Endsaldo nicht gefunden")
|
||
period_start = parse_german_date(opening.group(1))
|
||
period_end = parse_german_date(closing.group(1))
|
||
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
|
||
# 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:
|
||
txs.append(_finish(pending))
|
||
booking = parse_german_date(m.group(1))
|
||
pending = {
|
||
"booking": booking,
|
||
"value": booking,
|
||
"head": m.group(2).strip(),
|
||
"amount": parse_german_amount(m.group(3)),
|
||
"extra": [],
|
||
}
|
||
continue
|
||
if pending and not RE_OPENING.search(line) and not _is_noise(line):
|
||
pending["extra"].append(line)
|
||
if pending:
|
||
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(2)),
|
||
closing_balance=parse_german_amount(closing.group(2)),
|
||
transactions=txs)
|
||
|
||
|
||
def _finish(p: dict) -> ParsedTransaction:
|
||
if p["extra"]:
|
||
counterparty, remainder = _split_counterparty(p["extra"][0])
|
||
purpose_parts = [p["head"]]
|
||
if remainder:
|
||
purpose_parts.append(remainder)
|
||
purpose_parts.extend(p["extra"][1:])
|
||
else:
|
||
counterparty = ""
|
||
purpose_parts = [p["head"]]
|
||
purpose = " ".join(purpose_parts).strip()
|
||
return ParsedTransaction(booking_date=p["booking"], value_date=p["value"],
|
||
amount=p["amount"], purpose=purpose,
|
||
counterparty=counterparty)
|