import asyncio import logging from datetime import datetime, timedelta from apscheduler.schedulers.asyncio import AsyncIOScheduler from sqlalchemy.orm import Session from app.config import settings from app.database import SessionLocal from app.models.db_models import Account, FilterRule, LogLevel, ProcessedMail from app.services.encryption import decrypt from app.services.filter_engine import evaluate_conditions, execute_action from app.services.imap_client import CONNECTION_ERRORS, IMAPClient from app.services.log_service import write_log from app.services.maintenance import run_maintenance logger = logging.getLogger(__name__) scheduler = AsyncIOScheduler() # Nach so vielen geprüften Mails wird zwischengespeichert, damit der Poll die # SQLite-Schreibsperre nicht minutenlang hält (UI-Aktionen würden sonst warten). COMMIT_EVERY = 25 class ConnectionLost(Exception): """Die IMAP-Verbindung ist weg und ließ sich nicht wiederherstellen.""" def _reconnect(client: IMAPClient, reason: Exception | str) -> None: logger.warning("IMAP-Verbindung unterbrochen (%s) — verbinde neu", reason) client._current_folder = None try: client.reconnect() except Exception as e: raise ConnectionLost(f"{reason}; Neuverbindung fehlgeschlagen: {e}") from e def _fetch_with_reconnect(client: IMAPClient, folder: str, uid: str): """Mail holen; bei Verbindungsabbruch einmal neu verbinden und wiederholen. Ohne das scheiterten nach einem einzigen Abbruch alle restlichen Mails des Polls ('Broken pipe') und wurden trotzdem als verarbeitet markiert.""" for attempt in (1, 2): try: if getattr(client, "_current_folder", None) != folder: client.select_folder(folder) client._current_folder = folder return client.fetch_mail(uid) except CONNECTION_ERRORS as e: if attempt == 2: raise ConnectionLost(str(e)) from e _reconnect(client, e) def _verbose_log(**kwargs) -> None: """Routine-Einträge (Nicht-Übereinstimmung, 'keine neuen Mails', Poll-Start) erzeugten Millionen Logzeilen ohne Nutzen — nur bei LOG_VERBOSE schreiben.""" if settings.log_verbose: write_log(**kwargs) def _build_smtp_config(account: Account) -> dict | None: if account.smtp_host and account.smtp_username and account.smtp_password: return { "host": account.smtp_host, "port": account.smtp_port or 465, "username": account.smtp_username, "password": decrypt(account.smtp_password), } return None def _poll_account_sync(account_id: int) -> None: db: Session = SessionLocal() try: account = db.get(Account, account_id) if not account or not account.enabled: return rules = ( db.query(FilterRule) .filter(FilterRule.account_id == account_id, FilterRule.enabled.is_(True)) .order_by(FilterRule.priority) .all() ) _verbose_log( message=f"Poll gestartet ({len(rules)} aktive Regel(n))", level=LogLevel.INFO, account_id=account.id, account_name=account.name, details=", ".join(r.name for r in rules) if rules else "Keine Regeln konfiguriert", db=db, ) if not rules: account.last_poll_at = datetime.utcnow() db.commit() return source_folders = list({r.source_folder for r in rules}) smtp_config = _build_smtp_config(account) client = IMAPClient( host=account.imap_host, port=account.imap_port, username=account.username, password=decrypt(account.password), use_ssl=account.use_ssl, ) total_mails = 0 total_new = 0 total_matched = 0 total_actions = 0 total_errors = 0 evaluated_count = 0 # Mail-Cache: einmal geladene Mails pro Ordner/UID wiederverwenden mail_cache: dict[tuple[str, str], "MailMessage"] = {} # UIDs pro Ordner cachen folder_uids: dict[str, list[str]] = {} with client: # Phase 1: UIDs pro Ordner laden for folder in source_folders: try: all_uids = client.get_all_uids(folder, search="ALL") folder_uids[folder] = all_uids total_mails += len(all_uids) except Exception as e: write_log( message=f"Fehler beim Abrufen von Ordner '{folder}'", level=LogLevel.ERROR, account_id=account.id, account_name=account.name, folder=folder, details=str(e), db=db, ) # Phase 2: Pro Regel die unverarbeiteten Mails prüfen for rule in rules: folder = rule.source_folder if folder not in folder_uids: continue all_uids = folder_uids[folder] # Bereits verarbeitete UIDs für DIESE Regel processed_uids = set( row[0] for row in db.query(ProcessedMail.mail_uid) .filter( ProcessedMail.account_id == account.id, ProcessedMail.rule_id == rule.id, ProcessedMail.folder == folder, ) .all() ) # Neueste UIDs zuerst — sonst arbeitet der Scheduler bei großem # Backlog erst durch Jahre alte Newsletter, bevor er aktuelle # Rechnungen erreicht. UIDs sind auf IMAP-Servern monoton # steigend, daher sortiert reverse-int zuverlässig neu→alt. new_uids = [uid for uid in all_uids if uid not in processed_uids] try: new_uids.sort(key=int, reverse=True) except (ValueError, TypeError): new_uids.reverse() if not new_uids: _verbose_log( message=f"Regel '{rule.name}': keine neuen Mails in '{folder}' ({len(all_uids)} gesamt, alle bereits geprüft)", level=LogLevel.INFO, account_id=account.id, account_name=account.name, rule_name=rule.name, folder=folder, db=db, ) continue # Batch-Limit: maximal 500 Mails pro Regel pro Poll BATCH_LIMIT = 500 batch_uids = new_uids[:BATCH_LIMIT] remaining = len(new_uids) - len(batch_uids) total_new += len(batch_uids) msg = f"Regel '{rule.name}': {len(batch_uids)} Mail(s) in '{folder}' prüfen ({len(processed_uids)} bereits geprüft)" if remaining > 0: msg += f" — {remaining} weitere beim nächsten Poll" _verbose_log( message=msg, level=LogLevel.INFO, account_id=account.id, account_name=account.name, rule_name=rule.name, folder=folder, db=db, ) new_uids = batch_uids for uid in new_uids: # Mail aus Cache oder vom Server laden cache_key = (folder, uid) if cache_key in mail_cache: mail = mail_cache[cache_key] else: try: mail = _fetch_with_reconnect(client, folder, uid) except ConnectionLost: raise except Exception as e: write_log( message=f"Fehler beim Abrufen von Mail {uid}", level=LogLevel.ERROR, account_id=account.id, account_name=account.name, mail_uid=uid, folder=folder, details=str(e), db=db, ) # Trotzdem als verarbeitet markieren damit wir nicht endlos retrien db.add(ProcessedMail( account_id=account.id, rule_id=rule.id, folder=folder, mail_uid=uid, )) continue if not mail: db.add(ProcessedMail( account_id=account.id, rule_id=rule.id, folder=folder, mail_uid=uid, )) continue mail_cache[cache_key] = mail # Regel gegen Mail prüfen matched, details = evaluate_conditions(mail, rule.conditions) detail_str = " | ".join(details) if matched: total_matched += 1 rule.match_count = (rule.match_count or 0) + 1 rule.last_match_at = datetime.utcnow() # Aktionen ausführen retry_later = False for action in rule.actions: success = execute_action( client, mail, action, smtp_config, source_folder=folder, ) action_label = action.action_type.value param = action.parameter or "" if param: action_label += f" → {param}" if success: total_actions += 1 write_log( message=f"Aktion ausgeführt: {action_label}", level=LogLevel.SUCCESS, account_id=account.id, account_name=account.name, rule_name=rule.name, action_type=action.action_type.value, mail_uid=mail.uid, mail_subject=mail.subject, mail_from=mail.from_addr, folder=folder, details=param, db=db, ) else: total_errors += 1 reason = client.last_error or "unbekannter Grund" if not client.is_alive(): # Verbindung weg: neu verbinden und die Mail beim # nächsten Poll erneut versuchen statt sie als # verarbeitet abzuhaken _reconnect(client, reason) retry_later = True reason += " (Verbindung unterbrochen — wird beim nächsten Poll erneut versucht)" write_log( message=f"Aktion fehlgeschlagen: {action_label} — {reason}"[:1000], level=LogLevel.ERROR, account_id=account.id, account_name=account.name, rule_name=rule.name, action_type=action.action_type.value, mail_uid=mail.uid, mail_subject=mail.subject, mail_from=mail.from_addr, folder=folder, details=param, db=db, ) else: _verbose_log( message="Keine Übereinstimmung", level=LogLevel.INFO, account_id=account.id, account_name=account.name, rule_name=rule.name, mail_uid=mail.uid, mail_subject=mail.subject, mail_from=mail.from_addr, folder=folder, details=detail_str, db=db, ) if matched and retry_later: continue # Mail für DIESE Regel als verarbeitet markieren db.add(ProcessedMail( account_id=account.id, rule_id=rule.id, folder=folder, mail_uid=mail.uid, mail_subject=mail.subject[:500] if mail.subject else None, mail_from=mail.from_addr[:255] if mail.from_addr else None, )) evaluated_count += 1 if evaluated_count % COMMIT_EVERY == 0: db.commit() db.commit() # Poll-Zusammenfassung summary_parts = [ f"{total_mails} Mail(s) im Ordner", f"{total_new} neu", f"{total_matched} Treffer", f"{total_actions} Aktion(en)", ] if total_errors > 0: summary_parts.append(f"{total_errors} Fehler") # Nur loggen, wenn etwas passiert ist — ein Eintrag alle 30 s ohne # Inhalt verdeckt die relevanten Zeilen. Letzter Poll steht im Dashboard. summary_log = write_log if (total_new or total_actions or total_errors) else _verbose_log summary_log( message=f"Poll abgeschlossen: {', '.join(summary_parts)}", level=LogLevel.ERROR if total_errors > 0 else LogLevel.SUCCESS if total_actions > 0 else LogLevel.INFO, account_id=account.id, account_name=account.name, db=db, ) account.last_poll_at = datetime.utcnow() db.commit() except ConnectionLost as e: # Bisherigen Fortschritt behalten; die restlichen Mails sind nicht als # verarbeitet markiert und werden beim nächsten Poll geprüft. logger.error("IMAP-Verbindung für Konto %s verloren: %s", account_id, e) try: db.commit() except Exception: db.rollback() write_log( message="Verbindung zum Mailserver unterbrochen — Poll abgebrochen, restliche Mails werden beim nächsten Poll geprüft", level=LogLevel.WARNING, account_id=account_id, details=str(e), ) except Exception as e: logger.error("Fehler beim Polling von Konto %s: %s", account_id, e) db.rollback() write_log( message=f"Polling fehlgeschlagen", level=LogLevel.ERROR, account_id=account_id, details=str(e), ) finally: db.close() async def poll_account(account_id: int) -> None: await asyncio.to_thread(_poll_account_sync, account_id) async def maintenance_job() -> None: try: await asyncio.to_thread(run_maintenance) except Exception: pass # bereits in run_maintenance geloggt def add_account_job(account: Account) -> None: job_id = f"poll_account_{account.id}" if scheduler.get_job(job_id): scheduler.remove_job(job_id) if account.enabled: scheduler.add_job( poll_account, "interval", seconds=account.poll_interval_seconds, id=job_id, args=[account.id], replace_existing=True, ) logger.info( "Job für Konto '%s' registriert (alle %ds)", account.name, account.poll_interval_seconds, ) def remove_account_job(account_id: int) -> None: job_id = f"poll_account_{account_id}" if scheduler.get_job(job_id): scheduler.remove_job(job_id) logger.info("Job für Konto %s entfernt", account_id) def start_scheduler() -> None: db = SessionLocal() try: accounts = db.query(Account).filter(Account.enabled.is_(True)).all() for account in accounts: add_account_job(account) finally: db.close() # DB-Wartung (Log-Rotation, VACUUM): täglich zu fester Uhrzeit — ein # 24h-Intervall ab Start lief bei häufigen Deploys praktisch nie — und # zusätzlich kurz nach jedem Start. scheduler.add_job( maintenance_job, "cron", hour=settings.maintenance_hour_utc, minute=30, timezone="UTC", id="db_maintenance", replace_existing=True, ) scheduler.add_job( maintenance_job, "date", run_date=datetime.now() + timedelta(minutes=5), id="db_maintenance_startup", replace_existing=True, ) scheduler.start() logger.info("Scheduler gestartet mit %d Jobs", len(scheduler.get_jobs())) def stop_scheduler() -> None: scheduler.shutdown(wait=False) logger.info("Scheduler gestoppt")