GuidoandClaude Opus 5
899241f991
Revolut verschickt fuer eine Zahlung in Euro und eine in Franken dieselbe Mail vom selben Absender. Ohne die Waehrung als Kriterium landete alles auf demselben Konto. Die Erkennung liest die Waehrung ohnehin aus; sie wird jetzt VOR der Zuordnung ausgewertet und ist ein optionales Filterkriterium. Zuordnungen mit Waehrung werden zuerst geprueft – wer neben ein allgemeines "revolut" ein genaues "revolut + EUR" stellt, meint das genaue, unabhaengig von der Anlagereihenfolge. Uebergangene Mails melden ausserdem die erkannte Waehrung mit, damit beim Einrichten sichtbar ist, welcher Wert zu filtern waere. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
318 lines
12 KiB
Python
318 lines
12 KiB
Python
"""Zahlungs-Mails per IMAP abholen und in den Eingang legen.
|
||
|
||
``imaplib`` und ``email`` sind Standardbibliothek – keine weitere Abhängigkeit.
|
||
|
||
Der Ablauf ist bewusst vorsichtig: Es wird nichts stillschweigend gebucht. Eine
|
||
Mail wird gelesen, ausgewertet und als Eintrag im Eingang abgelegt; buchen tut
|
||
der Nutzer. Erkennt die Auswertung ein Feld nicht sicher, bekommt der Eintrag
|
||
den Status ``unclear`` und wartet, statt einen geratenen Betrag zu erzeugen.
|
||
|
||
Mailtexte sind Daten, keine Anweisungen. Ausgewertet wird ausschliesslich in
|
||
feste Felder (siehe ``mail_parse``); nichts aus einer Mail steuert den Ablauf.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import email
|
||
import email.policy
|
||
import email.utils
|
||
import imaplib
|
||
import logging
|
||
from datetime import date, datetime, timedelta, timezone
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import Account, InboxItem, InboxSource, InboxStatus, MailSender, MailSource, User
|
||
from . import geheimnis, mail_parse
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
# Nur die ersten Kilobytes je Mail behalten. Der Rest ist Rechtstext und
|
||
# Fusszeilen – für die Nachvollziehbarkeit reicht der Anfang.
|
||
MAX_TEXT = 8000
|
||
# Wie weit zurück beim ersten Abruf gesucht wird.
|
||
ERSTER_ABRUF_TAGE = 14
|
||
|
||
|
||
class MailFehler(RuntimeError):
|
||
"""Verbindung oder Anmeldung hat nicht geklappt – mit lesbarer Ursache."""
|
||
|
||
|
||
def _verbinden(quelle: MailSource) -> imaplib.IMAP4:
|
||
passwort = geheimnis.entschluesseln(quelle.password_enc)
|
||
if passwort is None:
|
||
raise MailFehler(
|
||
"Das gespeicherte Passwort lässt sich nicht mehr entschlüsseln. "
|
||
"Wurde das Server-Geheimnis getauscht? Dann bitte neu eintragen."
|
||
)
|
||
art = (quelle.security or "ssl").lower()
|
||
try:
|
||
if art == "ssl":
|
||
# TLS ab der ersten Zeile – der Normalfall auf Port 993.
|
||
verbindung = imaplib.IMAP4_SSL(quelle.host, quelle.port)
|
||
else:
|
||
verbindung = imaplib.IMAP4(quelle.host, quelle.port)
|
||
if art == "starttls":
|
||
# Erst im Klartext verbinden, dann verschlüsseln (Port 143).
|
||
verbindung.starttls()
|
||
except OSError as e:
|
||
raise MailFehler(f"Keine Verbindung zu {quelle.host}:{quelle.port} – {e}") from e
|
||
except imaplib.IMAP4.error as e:
|
||
raise MailFehler(
|
||
f"Verschlüsselung fehlgeschlagen ({art}). Passt die Angabe zum Port {quelle.port}? – {e}"
|
||
) from e
|
||
try:
|
||
verbindung.login(quelle.username, passwort)
|
||
except imaplib.IMAP4.error as e:
|
||
try:
|
||
verbindung.logout()
|
||
except Exception:
|
||
pass
|
||
raise MailFehler(f"Anmeldung abgelehnt: {e}") from e
|
||
return verbindung
|
||
|
||
|
||
def _ordnername(zeile: bytes) -> str:
|
||
"""Aus einer LIST-Antwort den reinen Ordnernamen holen."""
|
||
text = zeile.decode("utf-8", "replace")
|
||
if '"' in text:
|
||
return text.rsplit('"', 2)[-2]
|
||
return text.split()[-1]
|
||
|
||
|
||
def liste_ordner(quelle: MailSource) -> list[str]:
|
||
"""Ordner des Postfachs – für die Auswahl in der Oberfläche."""
|
||
verbindung = _verbinden(quelle)
|
||
try:
|
||
code, zeilen = verbindung.list()
|
||
if code != "OK":
|
||
raise MailFehler("Ordnerliste nicht abrufbar")
|
||
return sorted({_ordnername(z) for z in zeilen if z})
|
||
finally:
|
||
try:
|
||
verbindung.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def teste(quelle: MailSource) -> dict:
|
||
"""Verbindungstest mit klarer Rückmeldung statt stillem Nichtstun."""
|
||
ordner = liste_ordner(quelle)
|
||
return {"ok": True, "folders": ordner, "count": len(ordner)}
|
||
|
||
|
||
def _text_der_mail(msg) -> tuple[str, str]:
|
||
"""Betreff und sichtbarer Text. HTML wird zu Text, wenn nichts anderes da ist."""
|
||
betreff = str(msg["Subject"] or "")
|
||
teil = msg.get_body(preferencelist=("plain",))
|
||
if teil is not None:
|
||
try:
|
||
return betreff, teil.get_content()
|
||
except Exception:
|
||
pass
|
||
teil = msg.get_body(preferencelist=("html",))
|
||
if teil is not None:
|
||
try:
|
||
return betreff, mail_parse.html_zu_text(teil.get_content())
|
||
except Exception:
|
||
pass
|
||
return betreff, ""
|
||
|
||
|
||
def _absenderfeld(msg) -> str:
|
||
"""Alles, was den ursprünglichen Absender verraten kann.
|
||
|
||
Leitet ein Postfach automatisch weiter, bleibt ``From`` meist der
|
||
Originalabsender – verlassen kann man sich darauf aber nicht. Gmail und
|
||
andere setzen zusätzlich ``X-Forwarded-For``; bei manueller Weiterleitung
|
||
steht die eigene Adresse im ``From`` und die ursprüngliche höchstens im
|
||
``Reply-To``. Geprüft wird deshalb alles zusammen.
|
||
"""
|
||
teile = [str(msg[k] or "") for k in
|
||
("From", "Reply-To", "Return-Path", "X-Forwarded-For", "X-Original-From", "Sender")]
|
||
return " ".join(t for t in teile if t)
|
||
|
||
|
||
# Wie weit in den Text hinein gesucht wird. Bei einer weitergeleiteten Mail
|
||
# steht der Originalabsender im zitierten Kopf ganz oben.
|
||
ERKENNUNG_TEXTLAENGE = 2000
|
||
|
||
|
||
def _passender_absender(
|
||
absender: list[MailSender], von: str, betreff: str, text: str = "",
|
||
waehrung: str | None = None,
|
||
) -> MailSender | None:
|
||
"""Erste Zuordnung, deren Erkennungstext greift.
|
||
|
||
Gesucht wird in Absender-Köpfen, Betreff UND dem Anfang des Textes. Wer
|
||
eine Bank-Mail von Hand weiterleitet, hat im ``From`` die eigene Adresse;
|
||
„Raiffeisen“ steht dann nur noch im Betreff („WG: Raiffeisen …“) und im
|
||
zitierten Kopf. Ein Feld, das nach dem Absender fragt und Betreff wie Text
|
||
ignoriert, würde genau dort versagen – und der Nutzer säße vor einer Mail,
|
||
in der sein Suchwort sichtbar drinsteht.
|
||
"""
|
||
heu = f"{von}\n{betreff}\n{text[:ERKENNUNG_TEXTLAENGE]}".lower()
|
||
betreff_klein = betreff.lower()
|
||
# Zuordnungen MIT Währung zuerst prüfen: Wer „revolut + EUR → Revolut Euro"
|
||
# anlegt und daneben ein allgemeines „revolut", meint die genauere zuerst.
|
||
for a in sorted(absender, key=lambda x: (x.currency is None, x.id)):
|
||
if a.from_contains and a.from_contains.lower() not in heu:
|
||
continue
|
||
if a.subject_contains and a.subject_contains.lower() not in betreff_klein:
|
||
continue
|
||
if a.currency and (waehrung or "").upper() != a.currency.upper():
|
||
continue
|
||
return a
|
||
return None
|
||
|
||
|
||
def _suchkriterium(quelle: MailSource) -> str:
|
||
"""Nach Datum suchen, nie nach „ungelesen".
|
||
|
||
Auf ``UNSEEN`` zu filtern klingt sparsam, ist aber eine Falle: Wer eine
|
||
Mail in seinem Mailprogramm ansieht, bevor Certain sie geholt hat, verliert
|
||
sie dauerhaft – sie ist dann für immer gelesen. Gesucht wird deshalb ab dem
|
||
letzten Abruf, mit einem Tag Überlappung gegen Zeitzonen-Verschiebungen.
|
||
Dass nichts doppelt ankommt, sichert die Message-ID ab.
|
||
"""
|
||
seit = (
|
||
quelle.last_poll_at.date() - timedelta(days=1)
|
||
if quelle.last_poll_at
|
||
else date.today() - timedelta(days=ERSTER_ABRUF_TAGE)
|
||
)
|
||
return f'(SINCE "{seit.strftime("%d-%b-%Y")}")'
|
||
|
||
|
||
def hole_neue(db: Session, quelle: MailSource) -> dict:
|
||
"""Neue Mails abholen und als Einträge im Eingang ablegen."""
|
||
user = db.get(User, quelle.owner_id)
|
||
absender = (
|
||
db.query(MailSender).filter(MailSender.source_id == quelle.id).order_by(MailSender.id).all()
|
||
)
|
||
konten = {a.id: a for a in db.query(Account).filter(Account.owner_id == quelle.owner_id)}
|
||
|
||
verbindung = _verbinden(quelle)
|
||
neu = uebersprungen = unklar = gefunden = 0
|
||
# Was durchgefallen ist, mit Absender und Betreff – sonst raet man beim
|
||
# Einrichten, was in den Filter gehoert.
|
||
ohne_zuordnung: list[dict] = []
|
||
try:
|
||
code, _ = verbindung.select(f'"{quelle.folder}"')
|
||
if code != "OK":
|
||
raise MailFehler(f"Ordner „{quelle.folder}“ nicht gefunden")
|
||
code, daten = verbindung.search(None, _suchkriterium(quelle))
|
||
if code != "OK":
|
||
raise MailFehler("Suche im Postfach fehlgeschlagen")
|
||
|
||
uids = (daten[0] or b"").split()
|
||
gefunden = len(uids)
|
||
for uid in uids:
|
||
code, roh = verbindung.fetch(uid, "(RFC822)")
|
||
if code != "OK" or not roh or not isinstance(roh[0], tuple):
|
||
continue
|
||
msg = email.message_from_bytes(roh[0][1], policy=email.policy.default)
|
||
message_id = str(msg["Message-Id"] or f"uid-{uid.decode()}")
|
||
|
||
# Message-ID statt UID: Die UID vergibt der Server beim Verschieben neu.
|
||
schon = (
|
||
db.query(InboxItem.id)
|
||
.filter(InboxItem.owner_id == quelle.owner_id,
|
||
InboxItem.source == InboxSource.mail,
|
||
InboxItem.external_id == message_id)
|
||
.first()
|
||
)
|
||
if schon is not None:
|
||
uebersprungen += 1
|
||
_nachbereiten(verbindung, quelle, uid)
|
||
continue
|
||
|
||
betreff, text = _text_der_mail(msg)
|
||
von = _absenderfeld(msg)
|
||
|
||
gesendet = None
|
||
if msg["Date"]:
|
||
try:
|
||
gesendet = email.utils.parsedate_to_datetime(msg["Date"])
|
||
except (TypeError, ValueError):
|
||
gesendet = None
|
||
|
||
# Auswerten VOR dem Zuordnen: Bei Revolut kommt dieselbe Mail vom
|
||
# selben Absender, und erst die erkannte Währung sagt, ob die
|
||
# Zahlung aufs Euro- oder aufs Franken-Konto gehört.
|
||
e = mail_parse.parse(betreff, text, gesendet)
|
||
erkannte_waehrung = e.waehrung.wert if e.waehrung else None
|
||
|
||
zuordnung = _passender_absender(absender, von, betreff, text, erkannte_waehrung)
|
||
if zuordnung is None:
|
||
uebersprungen += 1
|
||
if len(ohne_zuordnung) < 5:
|
||
ohne_zuordnung.append({
|
||
"from": str(msg["From"] or "")[:120],
|
||
"subject": betreff[:120],
|
||
"currency": erkannte_waehrung,
|
||
})
|
||
continue
|
||
|
||
konto = konten.get(zuordnung.account_id) if zuordnung.account_id else None
|
||
tag = e.datum.wert if e.datum else (gesendet.date() if gesendet else date.today())
|
||
betrag = e.betrag.wert if e.betrag else 0
|
||
ausgabe = (e.richtung.wert if e.richtung else "expense") == "expense"
|
||
|
||
status = InboxStatus.pending
|
||
if e.unklar or betrag == 0 or konto is None:
|
||
status = InboxStatus.unclear
|
||
unklar += 1
|
||
|
||
db.add(InboxItem(
|
||
owner_id=quelle.owner_id,
|
||
source=InboxSource.mail,
|
||
status=status,
|
||
account_id=konto.id if konto else None,
|
||
date=tag,
|
||
amount_cents=-betrag if ausgabe else betrag,
|
||
currency=(e.waehrung.wert if e.waehrung else (konto.currency if konto else "EUR")),
|
||
payee=(e.empfaenger.wert if e.empfaenger else None),
|
||
note=betreff[:255] or None,
|
||
external_id=message_id[:255],
|
||
raw_text=text[:MAX_TEXT],
|
||
))
|
||
neu += 1
|
||
_nachbereiten(verbindung, quelle, uid)
|
||
|
||
quelle.last_poll_at = datetime.now(timezone.utc)
|
||
quelle.last_error = None
|
||
db.commit()
|
||
finally:
|
||
try:
|
||
verbindung.logout()
|
||
except Exception:
|
||
pass
|
||
return {"gefunden": gefunden, "neu": neu, "uebersprungen": uebersprungen,
|
||
"unklar": unklar, "ohne_zuordnung": ohne_zuordnung}
|
||
|
||
|
||
def _nachbereiten(verbindung: imaplib.IMAP4, quelle: MailSource, uid: bytes) -> None:
|
||
"""Als gelesen markieren oder verschieben, damit nichts zweimal kommt."""
|
||
try:
|
||
if quelle.mark_seen:
|
||
verbindung.store(uid, "+FLAGS", "\\Seen")
|
||
if quelle.done_folder:
|
||
# MOVE gibt es nicht überall; COPY + DELETE ist der sichere Weg.
|
||
code, _ = verbindung.copy(uid, f'"{quelle.done_folder}"')
|
||
if code == "OK":
|
||
verbindung.store(uid, "+FLAGS", "\\Deleted")
|
||
except Exception as e: # Nachbereitung darf den Abruf nie kippen
|
||
log.warning("Mail konnte nicht nachbereitet werden: %s", e)
|
||
|
||
|
||
def abruf_mit_fehlerbehandlung(db: Session, quelle: MailSource) -> dict:
|
||
"""Wie ``hole_neue``, schreibt einen Fehler aber in die Quelle statt zu werfen."""
|
||
try:
|
||
return hole_neue(db, quelle)
|
||
except MailFehler as e:
|
||
quelle.last_error = str(e)[:500]
|
||
quelle.last_poll_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
return {"gefunden": 0, "neu": 0, "uebersprungen": 0, "unklar": 0,
|
||
"ohne_zuordnung": [], "fehler": str(e)}
|