"Fehlgeschlagene neu einlesen" reported the same 29 mails on every click: it selected all error log entries ever written and never noted that one had already been retried. Error log entries now get retried_at when they are picked up; the endpoint only considers entries without it. If the mail fails again, the scheduler writes a new, open error entry, so a second click right after the first reports 0. The log shows retried errors greyed out with "↻ erneut versucht <Zeit>", and the result dialog says how many open error entries and mails were requeued. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
271 lines
9.7 KiB
Python
271 lines
9.7 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
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 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.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"}
|