"""Plugin „archiv“ — automatische Archivierung nach dem Sicherungsdatei- Prinzip. Titel der Neuheitenliste und der Planungsliste, deren Erscheinungsdatum bzw. Eintragsdatum länger als 12 Monate zurückliegt, werden vollständig in die Archiv-Tabellen (`archiv_neuheiten`, `archiv_planung`) verschoben und aus den aktiven Listen entfernt. Nichts wird gelöscht oder verändert — Admins können jeden Eintrag über die Archiv-Ansicht jederzeit wieder in die aktive Liste zurückverschieben. Umfang: - Eigene Migration mit beiden Spiegel-Tabellen. - Täglicher Hintergrund-Job (APScheduler, Cron um „03:00“ lokal). - Admin-Ansicht `/archiv` mit Suche, Herkunftsfilter und Wiederherstellen. - Je ein Audit-Log-Eintrag pro Archivierung (Akteur „System“) und pro Wiederherstellung (Akteur der Aktion). Konfiguration über Umgebungsvariablen (gelesen beim Plugin-Start): - SPIELE_ARCHIV_JOB_AKTIV „1“ (Standard) = täglicher Job an, „0“ = aus - SPIELE_ARCHIV_JOB_UHRZEIT Tageszeit des Laufs im Format HH:MM (Standard 03:00) Die Fachlogik liegt im Service (`service.py`) mit eingefrierbarer Uhr; Partner-Tabellen werden wie in dedup/planung nur über die gemeinsame SQLAlchemy-Metadata angesprochen. Der Kern bleibt unberührt. """ from __future__ import annotations import logging import os from urllib.parse import quote from fastapi import Depends, Request from fastapi.responses import RedirectResponse from sqlalchemy import select from sqlalchemy.orm import Session from redaktionskern.auth.deps import get_db, require_roles from redaktionskern.auth.models import Role, User from redaktionskern.contracts import BasePlugin, Migration, NavEntry from .models import STATUS_ANZEIGE, ArchivNeuheit, ArchivPlanung from .service import ( GUELTIGE_QUELLEN, QUELLE_NEUHEITEN, OBJEKT_TYPEN, ArchivService, ) _logger = logging.getLogger("plugins.archiv") JOB_ID = "archiv-taeglich" STANDARD_UHRZEIT = "03:00" def _uhrzeit_parsen(roh: str | None) -> tuple[int, int]: """Parst „HH:MM“; bei Unsinn gilt der Standard 03:00.""" teile = (roh or "").strip().split(":") try: stunde = int(teile[0]) minute = int(teile[1]) if len(teile) > 1 else 0 except (ValueError, IndexError): return 3, 0 if not (0 <= stunde <= 23 and 0 <= minute <= 59): return 3, 0 return stunde, minute def _tabelle_anlegen(conn) -> None: """Migration 0001: legt beide Spiegel-Tabellen an (portabel, idempotent).""" ArchivNeuheit.__table__.create(conn, checkfirst=True) ArchivPlanung.__table__.create(conn, checkfirst=True) def _spalte_existiert(conn, tabelle: str, spalte: str) -> bool: """Dialekt-agnostische Spaltenprüfung (PRAGMA bzw. information_schema).""" if conn.dialect.name == "sqlite": info = conn.exec_driver_sql(f"PRAGMA table_info({tabelle})").fetchall() return any(zeile[1] == spalte for zeile in info) existiert = conn.exec_driver_sql( "SELECT column_name FROM information_schema.columns " "WHERE table_name = :tabelle AND column_name = :spalte", {"tabelle": tabelle, "spalte": spalte}, ).scalar() return existiert is not None def _bild_url_spalten(conn) -> None: """Migration 0002: Spalte bild_url in beiden Spiegel-Tabellen. `ALTER TABLE … ADD COLUMN` mit nullable VARCHAR ohne Default ist SQLite- und Postgres-portabel und erhält bestehende Zeilen. """ for tabelle in ("archiv_neuheiten", "archiv_planung"): if _spalte_existiert(conn, tabelle, "bild_url"): continue conn.exec_driver_sql( f"ALTER TABLE {tabelle} ADD COLUMN bild_url VARCHAR(500)" ) class ArchivPlugin(BasePlugin): name = "archiv" title = "Archiv" description = ( "Verschiebt Titel älter als 12 Monate automatisch aus den aktiven " "Listen in das Archiv — vollständig wiederherstellbar." ) def __init__(self) -> None: super().__init__() self._scheduler = None # BackgroundScheduler, falls aktiviert #: Einfrierbare Uhr für Tests (Callable → aware datetime); None = echte Zeit. self._jetzt = None self._routen_registrieren() # ---------- Plugin-Vertrag ---------- def migrations(self) -> list[Migration]: return [ Migration(version="0001_archiv_tabellen", up=_tabelle_anlegen), Migration(version="0002_bild_url", up=_bild_url_spalten), ] def navigation(self) -> list[NavEntry]: return [NavEntry(label=self.title, url="/archiv")] def on_load(self, context) -> None: super().on_load(context) if os.environ.get("SPIELE_ARCHIV_JOB_AKTIV", "1").strip() == "1": self._scheduler_starten( os.environ.get("SPIELE_ARCHIV_JOB_UHRZEIT", STANDARD_UHRZEIT) ) def on_unload(self) -> None: if self._scheduler is not None: self._scheduler.shutdown(wait=False) self._scheduler = None super().on_unload() # ---------- Internas ---------- def _neuer_service(self) -> ArchivService: """Fabrik für den Service; die Uhr ist von Tests überschreibbar.""" return ArchivService(self.context.session_factory, jetzt=self._jetzt) def _audit_plugin(self): registry = getattr(self.context, "registry", None) return registry.get("audit-log") if registry is not None else None def _auditiere_archivierung(self, ereignisse) -> int: """Schreibt je archiviertem Titel einen Audit-Eintrag (Akteur: System).""" audit = self._audit_plugin() if audit is None: return 0 geschrieben = 0 for ereignis in ereignisse: try: audit.log_sync( None, "verschoben", ereignis.objekt_typ, ereignis.objekt_id, ereignis.details, ) geschrieben += 1 except Exception: # noqa: BLE001 - Audit darf den Lauf nicht brechen _logger.exception("Audit-Log-Eintrag für Archivierung fehlgeschlagen.") return geschrieben def _archiv_job(self) -> None: """Hintergrund-Job: läuft täglich und archiviert fällige Titel.""" service = self._neuer_service() lauf, ereignisse = service.lauf() self._auditiere_archivierung(ereignisse) if lauf.gesamt: _logger.info("Täglicher Archiv-Lauf abgeschlossen: %s", lauf.als_text()) def _scheduler_starten(self, uhrzeit_roh: str | None) -> None: from apscheduler.schedulers.background import BackgroundScheduler stunde, minute = _uhrzeit_parsen(uhrzeit_roh) scheduler = BackgroundScheduler() scheduler.add_job( self._archiv_job, trigger="cron", hour=stunde, minute=minute, id=JOB_ID, replace_existing=True, ) scheduler.start() self._scheduler = scheduler _logger.info( "Automatische Archivierung aktiv: täglich um %02d:%02d Uhr.", stunde, minute, ) # ---------- Routen ---------- def _routen_registrieren(self) -> None: @self.router.get("/archiv") def ansicht( request: Request, db: Session = Depends(get_db), user: User = Depends(require_roles(Role.ADMIN.value)), q: str = "", quelle: str = "", meldung: str = "", fehler: str = "", ): """Archiv-Ansicht (nur Admin): Suche, Filter, Wiederherstellen.""" service = self._neuer_service() eintraege = service.ansicht(db) alle_neuheiten = sum(1 for e in eintraege if e["quelle"] == QUELLE_NEUHEITEN) alle_planung = len(eintraege) - alle_neuheiten such = q.strip().lower() if such: eintraege = [ e for e in eintraege if such in (e["titel"] or "").lower() or such in (e["verlag"] or "").lower() or such in (e["autor"] or "").lower() ] if quelle in GUELTIGE_QUELLEN: eintraege = [e for e in eintraege if e["quelle"] == quelle] namen = { benutzer.id: (benutzer.display_name or benutzer.username) for benutzer in db.scalars(select(User)).all() } return self.context.templates.TemplateResponse( request=request, name="archiv/index.html", context={ "user": user, "titel": self.title, "eintraege": eintraege, "namen": namen, "status_anzeige": STATUS_ANZEIGE, "q": q, "quelle_filter": quelle if quelle in GUELTIGE_QUELLEN else "", "anzahl": len(eintraege), "anzahl_neuheiten": alle_neuheiten, "anzahl_planung": alle_planung, "meldung": meldung[:300], "fehler": fehler[:300], "job_aktiv": self._scheduler is not None and self._scheduler.running, }, ) @self.router.post("/archiv/{quelle}/{archiv_id}/wiederherstellen") async def wiederherstellen( request: Request, quelle: str, archiv_id: int, db: Session = Depends(get_db), user: User = Depends(require_roles(Role.ADMIN.value)), ): """Stellt einen archivierten Eintrag in die aktive Liste zurück.""" ziel_url = "/archiv" if quelle not in GUELTIGE_QUELLEN: return RedirectResponse( f"{ziel_url}?fehler={quote('Unbekannte Herkunft.')}", status_code=303, ) service = self._neuer_service() ergebnis = service.wiederherstellen( db, quelle, archiv_id, ersatz_rezensent_id=user.id ) if not ergebnis.ok: return RedirectResponse( f"{ziel_url}?fehler={quote(ergebnis.meldung)}", status_code=303 ) details = { "titel": ergebnis.titel, "von": "archiv", "ziel": ergebnis.ziel, "archiv_id": archiv_id, } if ergebnis.rezensent_ersetzt: details["rezensent_ersetzt"] = True audit = request.app.state.registry.get("audit-log") if audit is not None: try: await audit.log( user, "verschoben", OBJEKT_TYPEN[quelle], ergebnis.neuer_id, details, ) except Exception: # noqa: BLE001 - Audit blockiert nicht _logger.exception( "Audit-Log-Eintrag für Wiederherstellung fehlgeschlagen." ) return RedirectResponse( f"{ziel_url}?meldung={quote(ergebnis.meldung)}", status_code=303 ) plugin = ArchivPlugin()