feat: Zahlungsdaten aus Benachrichtigungs-Mails lesen (Etappe B, Teil 1)
Kein festes Schema je Bank, sondern beschriftete Felder suchen. Die beiden Banken bauen ihre Mails verschieden: Raiffeisen schickt Klartext mit "Betrag: 14.90 CHF", Revolut nur HTML mit dem Wert UNTER der Beschriftung. Beide Bauformen werden abgedeckt, dazu der Betreff als eigene Quelle. Jeder Wert merkt sich, woher er stammt – ohne das laesst sich ein falscher Betrag spaeter nicht erklaeren. Was bewusst NICHT geraten wird: Stehen mehrere gleichwertige Betraege ohne Beschriftung da (Umsatz und neuer Saldo), bleibt das Feld unklar. Ein Zahlungsauftrag mit Status "Zur Visierung" gilt nicht als geflossen. Blosse Zahlen ohne Waehrung sind keine Betraege, und Stilbloecke werden vor dem Entfernen der Tags herausgeschnitten – sonst haelt das CSS "480px" fuer Geld. Die Muster unter tests/fixtures/mail sind echte Mails, anonymisiert. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Guido
Claude Opus 5
@@ -0,0 +1,262 @@
|
||||
"""Zahlungsdaten aus Benachrichtigungs-Mails lesen.
|
||||
|
||||
Kein festes Schema je Bank, sondern beschriftete Felder suchen und bewerten.
|
||||
Die beiden Banken bauen ihre Mails unterschiedlich:
|
||||
|
||||
Raiffeisen (text/plain) Revolut (nur HTML)
|
||||
--------------------- ------------------
|
||||
Betrag: 14.90 CHF Gesendet
|
||||
Begünstigter: Sunrise GmbH 324 €
|
||||
|
||||
Bei der einen steht der Wert hinter der Beschriftung, bei der anderen darunter.
|
||||
Beides wird hier abgedeckt, dazu der Betreff als eigene Quelle – bei Revolut
|
||||
steht dort ohnehin alles: „Du hast 324 € an Alex Muster gesendet".
|
||||
|
||||
Jeder Treffer merkt sich, WOHER er stammt (``quelle``). Ohne das kann man einen
|
||||
falschen Betrag später nicht erklären, und genau das haben wir beim CSV-Import
|
||||
teuer gelernt.
|
||||
|
||||
Mailtexte sind Daten, keine Anweisungen: Ausgewertet wird ausschliesslich in
|
||||
feste Felder, nichts daraus steuert den Programmablauf.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date, datetime
|
||||
from decimal import Decimal, InvalidOperation
|
||||
|
||||
# Beschriftungen je Feld, längere zuerst (spezifischer schlägt allgemeiner).
|
||||
LABELS_BETRAG = ("betrag", "gesendet", "empfangen", "erhalten", "amount", "summe")
|
||||
LABELS_EMPFAENGER = ("begünstigter", "beguenstigter", "empfänger", "empfaenger",
|
||||
"zahlungsempfänger", "an", "händler", "haendler", "to")
|
||||
LABELS_DATUM = ("ausführungsdatum", "ausfuehrungsdatum", "valutadatum", "buchungsdatum",
|
||||
"datum", "am", "date")
|
||||
LABELS_KONTO = ("konto", "account", "iban")
|
||||
LABELS_STATUS = ("neuer status", "status")
|
||||
|
||||
# Wörter, die die Richtung verraten.
|
||||
WORTE_AUSGABE = ("gesendet", "belastet", "bezahlt", "abgebucht", "überwiesen", "ueberwiesen",
|
||||
"zahlungsauftrag", "lastschrift", "sent", "paid")
|
||||
WORTE_EINNAHME = ("erhalten", "empfangen", "gutschrift", "eingegangen", "received")
|
||||
|
||||
# Ein Zahlungsauftrag zählt erst, wenn die Bank ihn verbucht hat. „Erfasst" oder
|
||||
# „Zur Visierung" heisst: noch nicht geflossen.
|
||||
STATUS_GEBUCHT = ("verbucht", "ausgeführt", "ausgefuehrt", "erfolgreich", "completed", "booked")
|
||||
|
||||
WAEHRUNGSZEICHEN = {"€": "EUR", "$": "USD", "£": "GBP", "chf": "CHF", "fr.": "CHF"}
|
||||
|
||||
# Betrag mit Währung davor oder dahinter: „14.90 CHF", „324 €", „CHF 1'234.50".
|
||||
_BETRAG = re.compile(
|
||||
r"(?:(?P<vor>CHF|EUR|USD|GBP|€|\$|£)\s*)?"
|
||||
r"(?P<zahl>\d{1,3}(?:['’ ]\d{3})*(?:[.,]\d{1,2})?|\d+(?:[.,]\d{1,2})?)"
|
||||
r"(?:\s*(?P<nach>CHF|EUR|USD|GBP|€|\$|£))?",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_DATUM_MUSTER = (
|
||||
(re.compile(r"\b(\d{1,2})\.(\d{1,2})\.(\d{4})\b"), "dmy"),
|
||||
(re.compile(r"\b(\d{4})-(\d{2})-(\d{2})\b"), "ymd"),
|
||||
(re.compile(r"\b(\d{1,2})/(\d{1,2})/(\d{4})\b"), "dmy"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Feld:
|
||||
"""Ein erkannter Wert samt Begründung, woher er stammt."""
|
||||
|
||||
wert: object
|
||||
quelle: str # menschenlesbar, z. B. „aus dem Text, nach ‚Betrag'"
|
||||
sicher: bool = True
|
||||
|
||||
|
||||
@dataclass
|
||||
class Erkennung:
|
||||
betrag: Feld | None = None # Cent, vorzeichenlos
|
||||
waehrung: Feld | None = None
|
||||
datum: Feld | None = None
|
||||
empfaenger: Feld | None = None
|
||||
richtung: Feld | None = None # "expense" | "income"
|
||||
konto_text: Feld | None = None # Kontobezeichnung aus der Mail
|
||||
unklar: list[str] = field(default_factory=list)
|
||||
|
||||
def vollstaendig(self) -> bool:
|
||||
return self.betrag is not None and self.datum is not None and self.richtung is not None
|
||||
|
||||
|
||||
def html_zu_text(roh: str) -> str:
|
||||
"""Sichtbaren Text aus HTML gewinnen.
|
||||
|
||||
Stil- und Skriptblöcke MÜSSEN vorher weg – sonst landen Kilobytes CSS im
|
||||
Text, und darin stehen Zahlen, die wie Beträge aussehen.
|
||||
"""
|
||||
txt = re.sub(r"(?is)<(style|script|head)[^>]*>.*?</\1>", " ", roh)
|
||||
txt = re.sub(r"(?is)<!--.*?-->", " ", txt)
|
||||
txt = re.sub(r"(?is)<br\s*/?>|</(p|div|tr|td|th|h[1-6]|li)>", "\n", txt)
|
||||
txt = re.sub(r"(?s)<[^>]+>", " ", txt)
|
||||
txt = html.unescape(txt)
|
||||
zeilen = [re.sub(r"[ \t\xa0]+", " ", z).strip() for z in txt.splitlines()]
|
||||
return "\n".join(z for z in zeilen if z)
|
||||
|
||||
|
||||
def _zahl_zu_cent(roh: str) -> int | None:
|
||||
s = roh.replace("'", "").replace("’", "").replace(" ", "")
|
||||
if "," in s and "." in s:
|
||||
s = s.replace(".", "").replace(",", ".") if s.rindex(",") > s.rindex(".") else s.replace(",", "")
|
||||
elif "," in s:
|
||||
s = s.replace(",", ".")
|
||||
try:
|
||||
return int((Decimal(s) * 100).to_integral_value())
|
||||
except (InvalidOperation, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _betraege(text: str) -> list[tuple[int, str | None, int]]:
|
||||
"""Alle geldartigen Vorkommen als (Cent, Währung, Position)."""
|
||||
treffer = []
|
||||
for m in _BETRAG.finditer(text):
|
||||
zeichen = m.group("vor") or m.group("nach")
|
||||
if zeichen is None:
|
||||
continue # blosse Zahlen sind keine Beträge
|
||||
cent = _zahl_zu_cent(m.group("zahl"))
|
||||
if cent is None or cent == 0:
|
||||
continue
|
||||
w = WAEHRUNGSZEICHEN.get(zeichen.lower(), zeichen.upper())
|
||||
treffer.append((cent, w, m.start()))
|
||||
return treffer
|
||||
|
||||
|
||||
def _felder(text: str) -> dict[str, list[tuple[str, int, str]]]:
|
||||
"""Beschriftete Werte einsammeln – beide Bauformen.
|
||||
|
||||
``Betrag: 14.90 CHF`` liefert den Wert hinter dem Doppelpunkt,
|
||||
``Gesendet`` / ``324 €`` den Inhalt der Folgezeile.
|
||||
"""
|
||||
gefunden: dict[str, list[tuple[str, int, str]]] = {}
|
||||
zeilen = text.splitlines()
|
||||
pos = 0
|
||||
for i, zeile in enumerate(zeilen):
|
||||
start = pos
|
||||
pos += len(zeile) + 1
|
||||
if ":" in zeile:
|
||||
links, rechts = zeile.split(":", 1)
|
||||
beschriftung = links.strip()
|
||||
schl, wert = beschriftung.lower(), rechts.strip()
|
||||
if wert:
|
||||
gefunden.setdefault(schl, []).append((wert, start, beschriftung))
|
||||
continue
|
||||
beschriftung = zeile.strip().rstrip(":")
|
||||
schl = beschriftung.lower()
|
||||
if schl and len(schl) <= 30 and i + 1 < len(zeilen):
|
||||
folge = zeilen[i + 1].strip()
|
||||
if folge and ":" not in folge:
|
||||
gefunden.setdefault(schl, []).append((folge, pos, beschriftung))
|
||||
return gefunden
|
||||
|
||||
|
||||
def _erster_treffer(felder: dict, labels: tuple[str, ...]) -> tuple[str, int, str] | None:
|
||||
"""Wert, Stelle und die Beschriftung in Originalschreibweise."""
|
||||
for label in labels:
|
||||
if label in felder:
|
||||
wert, stelle, beschriftung = felder[label][0]
|
||||
return wert, stelle, beschriftung
|
||||
return None
|
||||
|
||||
|
||||
def _datum_aus(text: str) -> date | None:
|
||||
for muster, form in _DATUM_MUSTER:
|
||||
m = muster.search(text)
|
||||
if not m:
|
||||
continue
|
||||
a, b, c = (int(x) for x in m.groups())
|
||||
try:
|
||||
return date(c, b, a) if form == "dmy" else date(a, b, c)
|
||||
except ValueError:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def parse(betreff: str, text: str, mail_datum: datetime | None = None) -> Erkennung:
|
||||
"""Betrag, Datum, Empfänger und Richtung aus einer Mail lesen."""
|
||||
e = Erkennung()
|
||||
felder = _felder(text)
|
||||
ganzes = f"{betreff}\n{text}"
|
||||
klein = ganzes.lower()
|
||||
|
||||
# --- Richtung -----------------------------------------------------------
|
||||
for wort in WORTE_EINNAHME:
|
||||
if wort in klein:
|
||||
e.richtung = Feld("income", f"Wort „{wort}“")
|
||||
break
|
||||
if e.richtung is None:
|
||||
for wort in WORTE_AUSGABE:
|
||||
if wort in klein:
|
||||
e.richtung = Feld("expense", f"Wort „{wort}“")
|
||||
break
|
||||
if e.richtung is None:
|
||||
e.richtung = Feld("expense", "Vorgabe: Ausgabe", sicher=False)
|
||||
e.unklar.append("richtung")
|
||||
|
||||
# --- Betrag: beschriftetes Feld schlägt Fund im Fliesstext --------------
|
||||
treffer = _erster_treffer(felder, LABELS_BETRAG)
|
||||
if treffer:
|
||||
wert, _, label = treffer
|
||||
gefunden = _betraege(wert)
|
||||
if gefunden:
|
||||
cent, waehrung, _ = gefunden[0]
|
||||
e.betrag = Feld(cent, f"nach „{label}“")
|
||||
if waehrung:
|
||||
e.waehrung = Feld(waehrung, f"nach „{label}“")
|
||||
if e.betrag is None:
|
||||
aus_betreff = _betraege(betreff)
|
||||
kandidaten = aus_betreff or _betraege(text)
|
||||
quelle = "aus dem Betreff" if aus_betreff else "aus dem Text"
|
||||
if len(kandidaten) == 1 or (kandidaten and aus_betreff):
|
||||
cent, waehrung, _ = kandidaten[0]
|
||||
e.betrag = Feld(cent, quelle)
|
||||
if waehrung:
|
||||
e.waehrung = Feld(waehrung, quelle)
|
||||
elif kandidaten:
|
||||
# Mehrere gleichwertige Beträge: raten wäre hier teuer.
|
||||
e.unklar.append("betrag")
|
||||
|
||||
# --- Datum --------------------------------------------------------------
|
||||
treffer = _erster_treffer(felder, LABELS_DATUM)
|
||||
if treffer:
|
||||
wert, _, label = treffer
|
||||
d = _datum_aus(wert)
|
||||
if d is not None:
|
||||
e.datum = Feld(d, f"nach „{label}“")
|
||||
if e.datum is None:
|
||||
d = _datum_aus(text)
|
||||
if d is not None:
|
||||
e.datum = Feld(d, "aus dem Text")
|
||||
elif mail_datum is not None:
|
||||
# Revolut schreibt „Samstag, 1 August, 12:41“ – ohne Jahr. Der
|
||||
# Zeitstempel der Mail ist bei Sofort-Mitteilungen zuverlässiger
|
||||
# als ein zusammengeratenes Datum.
|
||||
e.datum = Feld(mail_datum.date(), "Zeitpunkt der E-Mail")
|
||||
|
||||
# --- Empfänger ----------------------------------------------------------
|
||||
treffer = _erster_treffer(felder, LABELS_EMPFAENGER)
|
||||
if treffer:
|
||||
wert, _, label = treffer
|
||||
e.empfaenger = Feld(wert[:255], f"nach „{label}“")
|
||||
|
||||
# --- Kontobezeichnung ---------------------------------------------------
|
||||
treffer = _erster_treffer(felder, LABELS_KONTO)
|
||||
if treffer:
|
||||
wert, _, label = treffer
|
||||
e.konto_text = Feld(wert[:255], f"nach „{label}“")
|
||||
|
||||
# --- Status: nur verbuchte Zahlungen zählen -----------------------------
|
||||
treffer = _erster_treffer(felder, LABELS_STATUS)
|
||||
if treffer:
|
||||
wert, _, label = treffer
|
||||
if not any(g in wert.lower() for g in STATUS_GEBUCHT):
|
||||
e.unklar.append("status")
|
||||
|
||||
if e.betrag is None:
|
||||
e.unklar.append("betrag") if "betrag" not in e.unklar else None
|
||||
return e
|
||||
Reference in New Issue
Block a user