GuidoandClaude Opus 5
615be0f0f5
Bei einer von Hand weitergeleiteten Bank-Mail steht die eigene Adresse im
From; "Raiffeisen" findet sich nur noch im Betreff ("WG: Raiffeisen …")
und im zitierten Kopf. Das Feld hiess "Absender enthaelt" und prueft nur die
Absender-Koepfe – die Mail wurde also uebergangen, obwohl das Suchwort
sichtbar darin stand. Eine Falle, kein Bedienfehler.
Gesucht wird jetzt in Absender-Koepfen, Betreff und den ersten 2000 Zeichen
des Textes; die Beschriftung heisst entsprechend "Erkennungstext". Ein Test
mit einer nachgebauten Weiterleitung haelt fest, dass sowohl die Zuordnung
greift als auch Betrag und Empfaenger daraus lesbar bleiben.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
306 lines
12 KiB
Python
306 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 = ""
|
||
) -> 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()
|
||
for a in absender:
|
||
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
|
||
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)
|
||
zuordnung = _passender_absender(absender, von, betreff, text)
|
||
if zuordnung is None:
|
||
uebersprungen += 1
|
||
if len(ohne_zuordnung) < 5:
|
||
ohne_zuordnung.append({
|
||
"from": str(msg["From"] or "")[:120],
|
||
"subject": betreff[:120],
|
||
})
|
||
continue
|
||
|
||
gesendet = None
|
||
if msg["Date"]:
|
||
try:
|
||
gesendet = email.utils.parsedate_to_datetime(msg["Date"])
|
||
except (TypeError, ValueError):
|
||
gesendet = None
|
||
|
||
e = mail_parse.parse(betreff, text, gesendet)
|
||
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)}
|