Files
imap-mail-filter-service/app/routers/filters.py
T
duffyduckandClaude Opus 5.5 d5cb0a83f3 new action: attach the message as PDF and file it
For senders whose mail body is the receipt (e.g. GoCardless payouts)
there is no attachment to file. The new action "Als PDF anhängen +
nach <Ordner>" renders the mail like a print from a mail client
(header block From/To/Date/Subject, then the content) and replaces it
in the target folder with: original content + PDF + the original mail
byte-for-byte as Original.eml, so the unaltered original is retained
(GoBD).

- Rendering with WeasyPrint. Images are fetched by us and inlined as
  data: URIs; WeasyPrint itself may only read data:. Fetching is
  restricted to http(s), public addresses (private, loopback,
  link-local and PDF_BLOCKED_NETWORKS, default 172.0.0.0/12, are
  refused; redirects are checked hop by hop), with time and size
  limits. cid: images come from the MIME parts.
- The replacement is APPENDed with the original flags and INTERNALDATE,
  verified in the target via its new Message-ID, and only then is the
  original deleted. Already converted mails are only moved.
- "PDF-Vorschau" in the rule dialog renders the newest matching mail
  read-only and opens it in the browser.
- List/log show 📄 targets; warnings for PDF + move/delete and for
  PDF + "Hat Anhang = Ja"; log folder search covers PDF targets.
- Docker image gets Pango and fonts (~316 MB total).

Tested with the real GoCardless mail (3 remote images, rendered both
locally and inside the built image) and a simulated IMAP server for
success, rejected APPEND, missing copy and already converted mails.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 19:51:26 +02:00

298 lines
11 KiB
Python

from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import Response
from sqlalchemy import func, or_, text
from sqlalchemy.orm import Session
import logging
from app.database import engine, get_db
from app.models.db_models import (
Account,
FilterAction,
FilterCondition,
FilterLog,
FilterRule,
LogLevel,
ProcessedMail,
)
from app.schemas.schemas import (
FilterRuleCreate,
FilterRuleEnabled,
FilterRuleResponse,
FilterRuleTest,
FilterRuleUpdate,
)
from app.services.rule_tester import preview_pdf, test_rule
logger = logging.getLogger(__name__)
def _reset_processed_for_rule(db: Session, rule_id: int) -> int:
"""Reset processed mails for a specific rule so they get re-evaluated."""
count = (
db.query(ProcessedMail)
.filter(ProcessedMail.rule_id == rule_id)
.delete()
)
if count:
logger.info("Filter geändert: %d verarbeitete Mails für Regel %d zurückgesetzt", count, rule_id)
return count
router = APIRouter(prefix="/api/filters", tags=["filters"])
@router.post("/test")
def test_filter(data: FilterRuleTest, db: Session = Depends(get_db)):
"""Prüft eine Regel read-only gegen die neuesten passenden Mails im
Quellordner, ohne sie zu speichern oder Mails zu verändern."""
account = db.get(Account, data.account_id)
if not account:
raise HTTPException(404, "Konto nicht gefunden")
if not data.conditions:
raise HTTPException(400, "Keine Bedingungen angegeben")
try:
return test_rule(account, data.source_folder, data.conditions, data.limit)
except Exception as e:
logger.exception("Regeltest fehlgeschlagen")
raise HTTPException(500, f"Test fehlgeschlagen: {e}")
@router.post("/pdf-preview")
def pdf_preview(data: FilterRuleTest, db: Session = Depends(get_db)):
"""PDF-Vorschau der neuesten passenden Mail (für die Aktion 'Als PDF') —
read-only, es wird nichts verändert."""
account = db.get(Account, data.account_id)
if not account:
raise HTTPException(404, "Konto nicht gefunden")
if not data.conditions:
raise HTTPException(400, "Keine Bedingungen angegeben")
try:
pdf, info = preview_pdf(account, data.source_folder, data.conditions, data.limit)
except LookupError as e:
raise HTTPException(404, str(e))
except Exception as e:
logger.exception("PDF-Vorschau fehlgeschlagen")
raise HTTPException(500, f"PDF-Vorschau fehlgeschlagen: {e}")
return Response(
content=pdf,
media_type="application/pdf",
headers={
"Content-Disposition": 'inline; filename="vorschau.pdf"',
"X-Preview-Images": f"{info['loaded']} geladen, {info['skipped']} übersprungen, {info['cid']} eingebettet",
},
)
@router.patch("/{rule_id}/enabled", response_model=FilterRuleResponse)
def set_filter_enabled(rule_id: int, data: FilterRuleEnabled, db: Session = Depends(get_db)):
"""Regel ein-/ausschalten, ohne den Verarbeitungsstatus zurückzusetzen
(ein normales Update würde alle Mails für diese Regel neu prüfen lassen)."""
rule = db.get(FilterRule, rule_id)
if not rule:
raise HTTPException(404, "Filterregel nicht gefunden")
rule.enabled = data.enabled
db.commit()
db.refresh(rule)
return rule
@router.post("/reset-all-processed")
def reset_all_processed(db: Session = Depends(get_db)):
"""Setzt den 'verarbeitet'-Status ALLER Mails zurück — alle Regeln bewerten
beim nächsten Poll wieder jede Mail im Ordner neu. Sicher, weil schon
verschobene Mails nicht mehr in der Quelle liegen und daher nicht erneut
verschoben werden können. Nutzt DROP+CREATE statt DELETE, damit auch
bei Millionen Zeilen sekundenschnell."""
try:
row_count = db.query(ProcessedMail).count()
db.close() # Sitzung freigeben, damit DDL nicht am Lock hängt
with engine.begin() as conn:
ProcessedMail.__table__.drop(conn, checkfirst=True)
ProcessedMail.__table__.create(conn, checkfirst=True)
conn.execute(text("PRAGMA wal_checkpoint(TRUNCATE)"))
logger.info("ProcessedMail vollständig zurückgesetzt (%d Einträge entfernt)", row_count)
return {"reset": row_count}
except Exception as e:
logger.exception("reset-all-processed fehlgeschlagen")
raise HTTPException(500, f"Fehler beim Zurücksetzen: {e}")
@router.post("/retry-failed")
def retry_failed_fetches(account_id: int | None = None, db: Session = Depends(get_db)):
"""Stößt Mails, bei denen das Abrufen oder eine Aktion (z.B. Verschieben)
fehlgeschlagen ist, für den nächsten Poll erneut an: ihre 'verarbeitet'-
Marker werden gelöscht. Schon verschobene Mails liegen nicht mehr im
Quellordner und werden daher nicht doppelt verarbeitet.
Nur offene Fehler zählen: die aufgegriffenen Log-Einträge werden mit
retried_at markiert. Scheitert die Mail erneut, entsteht ein neuer,
wieder offener Fehler-Eintrag."""
try:
log_q = db.query(FilterLog.id, FilterLog.mail_uid, FilterLog.account_id).filter(
FilterLog.level == LogLevel.ERROR,
or_(
FilterLog.message.like("Fehler beim Abrufen%"),
FilterLog.message.like("Aktion fehlgeschlagen%"),
),
FilterLog.mail_uid.isnot(None),
FilterLog.retried_at.is_(None),
)
if account_id is not None:
log_q = log_q.filter(FilterLog.account_id == account_id)
rows = log_q.all()
if not rows:
return {"reset": 0, "unique_uids": 0, "log_entries": 0}
by_account: dict[int | None, set[str]] = {}
for _, uid, acc_id in rows:
by_account.setdefault(acc_id, set()).add(uid)
total = 0
for acc_id, uids in by_account.items():
q = db.query(ProcessedMail).filter(ProcessedMail.mail_uid.in_(list(uids)))
if acc_id is not None:
q = q.filter(ProcessedMail.account_id == acc_id)
total += q.delete(synchronize_session=False)
# Aufgegriffene Fehler als erneut versucht markieren (in Blöcken, SQLite
# begrenzt die Zahl der Parameter pro Anweisung)
now = datetime.utcnow()
ids = [row[0] for row in rows]
for i in range(0, len(ids), 500):
db.query(FilterLog).filter(FilterLog.id.in_(ids[i:i + 500])).update(
{FilterLog.retried_at: now}, synchronize_session=False
)
db.commit()
unique_uids = sum(len(v) for v in by_account.values())
logger.info(
"Retry-Failed: %d Fehler-Einträge, %d Mails, %d Marker entfernt",
len(ids), unique_uids, total,
)
return {"reset": total, "unique_uids": unique_uids, "log_entries": len(ids)}
except Exception as e:
logger.exception("retry-failed fehlgeschlagen")
raise HTTPException(500, f"Fehler bei retry-failed: {e}")
@router.get("/account/{account_id}", response_model=list[FilterRuleResponse])
def list_filters(account_id: int, db: Session = Depends(get_db)):
account = db.get(Account, account_id)
if not account:
raise HTTPException(404, "Konto nicht gefunden")
rules = (
db.query(FilterRule)
.filter(FilterRule.account_id == account_id)
.order_by(FilterRule.priority)
.all()
)
return rules
@router.get("/{rule_id}", response_model=FilterRuleResponse)
def get_filter(rule_id: int, db: Session = Depends(get_db)):
rule = db.get(FilterRule, rule_id)
if not rule:
raise HTTPException(404, "Filterregel nicht gefunden")
return rule
@router.post("/", response_model=FilterRuleResponse, status_code=201)
def create_filter(data: FilterRuleCreate, db: Session = Depends(get_db)):
account = db.get(Account, data.account_id)
if not account:
raise HTTPException(404, "Konto nicht gefunden")
if data.priority is None:
# Ans Ende einsortieren: max(priority) + 10 für dieses Konto. Falls noch
# keine Regel existiert, starten wir bei 10.
current_max = db.query(func.max(FilterRule.priority)).filter(
FilterRule.account_id == data.account_id
).scalar()
priority = (current_max + 10) if current_max is not None else 10
else:
priority = data.priority
rule = FilterRule(
account_id=data.account_id,
name=data.name,
priority=priority,
enabled=data.enabled,
stop_processing=data.stop_processing,
source_folder=data.source_folder,
)
db.add(rule)
db.flush()
for cond_data in data.conditions:
cond = FilterCondition(rule_id=rule.id, **cond_data.model_dump())
db.add(cond)
for action_data in data.actions:
action = FilterAction(rule_id=rule.id, **action_data.model_dump())
db.add(action)
# Neue Regel → hat noch keinen processed-Status, wird automatisch alle Mails prüfen
db.commit()
db.refresh(rule)
return rule
@router.put("/{rule_id}", response_model=FilterRuleResponse)
def update_filter(rule_id: int, data: FilterRuleUpdate, db: Session = Depends(get_db)):
rule = db.get(FilterRule, rule_id)
if not rule:
raise HTTPException(404, "Filterregel nicht gefunden")
update_data = data.model_dump(exclude_unset=True)
# Update conditions if provided
if "conditions" in update_data:
for cond in rule.conditions:
db.delete(cond)
for cond_data in data.conditions:
cond = FilterCondition(rule_id=rule.id, **cond_data.model_dump())
db.add(cond)
del update_data["conditions"]
# Update actions if provided
if "actions" in update_data:
for action in rule.actions:
db.delete(action)
for action_data in data.actions:
action = FilterAction(rule_id=rule.id, **action_data.model_dump())
db.add(action)
del update_data["actions"]
for key, value in update_data.items():
setattr(rule, key, value)
# Regel geändert → nur diese Regel zurücksetzen
_reset_processed_for_rule(db, rule.id)
db.commit()
db.refresh(rule)
return rule
@router.delete("/{rule_id}", status_code=204)
def delete_filter(rule_id: int, db: Session = Depends(get_db)):
rule = db.get(FilterRule, rule_id)
if not rule:
raise HTTPException(404, "Filterregel nicht gefunden")
# processed-Einträge werden per CASCADE gelöscht
db.delete(rule)
db.commit()
@router.put("/reorder/{account_id}")
def reorder_filters(account_id: int, rule_ids: list[int], db: Session = Depends(get_db)):
for priority, rule_id in enumerate(rule_ids):
rule = db.get(FilterRule, rule_id)
if rule and rule.account_id == account_id:
rule.priority = priority
db.commit()
return {"message": "Reihenfolge aktualisiert"}