b0rbor4d/spiele-redaktion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Plugin archiv: 12-Monats-Autopilot mit Sicherungsdatei-Prinzip

Browse Source 
- Eigene Migration 0001_archiv_tabellen: Spiegel-Tabellen archiv_neuheiten
  und archiv_planung (ohne FK, damit Archiv-Zeilen Benutzerlöschung überleben)
- Service mit eingefrierbarer Uhr: Stichtag „jetzt minus 12 Monate“ in UTC,
  strenger Vergleich (genau 12 Monate bleibt aktiv); Neuheiten nach
  Erscheinungsjahr (großzügig Jahresende) sonst Eintragsdatum, Planung nach
  Eintragsdatum; Monatsarithmetik mit Klemmung (29. Februar)
- Täglicher APScheduler-Cron-Job (Standard 03:00, konfigurierbar über
  SPIELE_ARCHIV_JOB_UHRZEIT / SPIELE_ARCHIV_JOB_AKTIV); je archiviertem
  Titel ein Audit-Log-Eintrag als System
- Admin-Ansicht /archiv: vereinigte Liste beider Tabellen, Suche über
  Titel/Verlag/Autor, Herkunftsfilter, Wiederherstellen in die aktive Liste
  (BGG-Konflikt wird abgelehnt, verwaiste Rezensentenzuordnung übernimmt der
  Admin — gemeldet und auditiert); Rollen: nur Admin
- Kern minimal erweitert: PluginContext.registry stellt Hintergrund-Jobs die
  Plugin-Registry bereit (keine Fachlogik im Kern)
- 30 neue Tests mit eingefrorener Uhr (Grenzfälle, Zeitzonen, Schaltjahr,
  Job-Lauf, Wiederherstellen, Rollen, Suche/Filter); 171 Tests grün
- README aktualisiert: neues Plugin-Kapitel, Env-Variablen, Fortschritt #7 ✅
This commit is contained in:
Flo Hartmann
2026-08-21 20:26:00 +00:00
parent 0f12cd952b
commit 72fe3b7eda
10 changed files with 1727 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
README.md
View File

@@ -4,7 +4,7 @@
modularer Plugin-Architektur. Lauffähiger Kern mit Plugin-System,
Authentifizierung/Rollen und Migrationen; vollständig implementiert sind
bisher **Audit-Log**, **Benachrichtigung**, **Neuheiten (BGG-Sync)**,
**Dedup-Prüfung** und **Planungsliste**.
**Dedup-Prüfung**, **Planungsliste** und **Archiv (12-Monats-Autopilot)**.
## Stack
@@ -54,6 +54,8 @@ Beim ersten Start wird automatisch ein Admin-Konto angelegt:
| `SPIELE_BGG_SUCHBEGRIFFE` | `brettspiel` | Komma-getrennte Suchbegriffe für den regelmäßigen Sync |
| `SPIELE_BGG_MAX_TREFFER_PRO_SUCHE` | `25` | Obergrenze Treffer je Suchbegriff (schont das BGG-Rate-Limit) |
| `SPIELE_DEDUP_BGG_AKTIV` | `1` | BGG-Zusatzdaten für die Dedup-Prüfung an (`1`) oder aus (`0`): Alternate-Names und Erweiterungs-Relationen |
| `SPIELE_ARCHIV_JOB_AKTIV` | `1` | Täglicher Archivierungs-Job an (`1`) oder aus (`0`) |
| `SPIELE_ARCHIV_JOB_UHRZEIT` | `03:00` | Tageszeit des täglichen Archiv-Laufs im Format `HH:MM` |
## Tests
@@ -75,6 +77,11 @@ Fake-BGG-Client, Prüfprotokoll, Prüfseite) und **planung** (beide Eintragswege
mit Stub-gemockter dedup-/Benachrichtigungs-/Audit-API, Verlags-Auswahl-
Dialog, Statuswechsel, Bearbeiten/Löschen, Rechte pro Rolle) plus ein
Integrationstest mit den echten Plugins (In-App-Nachricht + Audit-Eintrag).
Dazu das **archiv**-Plugin: 12-Monats-Regel mit eingefrorener Uhr (Grenzfälle
„genau 12 Monate“, Schaltjahr/29. Februar, UTC-Konsistenz bei
Zeitzonen-Unterschieden), Job-Lauf mit Audit-Einträgen, Wiederherstellen
inkl. BGG-Konflikt und Ersatz-Rezensent, Rollen sowie Suche/Filter in der
Admin-Ansicht.
## Architektur
@@ -94,7 +101,8 @@ plugins/ ein Ordner pro Funktion, ladbar über den Plugin-Load
├── benachrichtigung/ VOLL IMPLEMENTIERT: E-Mail/Telegram/In-App
├── dedup/ VOLL IMPLEMENTIERT: check_titel-API, Heuristik, Protokoll
├── planung/ VOLL IMPLEMENTIERT: Planungsliste, Verschiebung, Dialoge
├── archiv/ erinnerung/ export/ (Stubs, ladbar)
├── archiv/ VOLL IMPLEMENTIERT: 12-Monats-Autopilot, Job, UI
├── erinnerung/ export/ (Stubs, ladbar)
└── … je __init__.py + templates/<name>/
```
@@ -122,7 +130,9 @@ class MeinPlugin(BasePlugin):
return [NavEntry(label=self.title, url="/meinplugin")]
def on_load(self, context): # Lifecycle-Hook beim Start
super().on_load(context) # context.engine/.session_factory/.templates/.settings
super().on_load(context) # context.engine/.session_factory/.templates/
# .settings/.registry (Plugin-Registry für
# Hintergrund-Jobs ohne Request)
def on_unload(self): # Lifecycle-Hook beim Herunterfahren
super().on_unload()
@@ -363,6 +373,64 @@ Rezensenten können **nur ihre eigenen Einträge** bearbeiten (Statuswechsel,
Bearbeiten, Löschen); Admins und Redakteure alle. Jede Änderung wird im
Audit-Log protokolliert (`erstellt`/`verschoben`/`geaendert`/`geloescht`).
## Plugin „archiv“ (implementiert)
Automatische Archivierung nach dem **Sicherungsdatei-Prinzip**: Titel, deren
Erscheinungs- bzw. Eintragsdatum länger als 12 Monate zurückliegt, werden
vollständig in eigene Archiv-Tabellen verschoben und aus den aktiven Listen
entfernt. Nichts wird gelöscht oder verändert — jeder Eintrag lässt sich
unverändert zurück in die aktive Liste verschieben.
### Datentabellen
Eigene Migration `0001_archiv_tabellen` mit zwei Spiegel-Tabellen
(SQLite ↔ Postgres portabel, ohne Fremdschlüssel — archivierte Zeilen
überleben auch das Löschen eines Benutzers):
| Tabelle | Spiegel von | Besondere Spalten |
|---------|-------------|-------------------|
| `archiv_neuheiten` | `neuheiten` | alle Originalspalten plus `quell_id`, `archiviert_am`, `archiviert_von`, `grund`, `referenz_am` |
| `archiv_planung` | `planungsliste` | alle Originalspalten (inkl. `pruefung`-JSON) plus dieselben Archiv-Metadaten |
### Die 12-Monats-Regel
- **Stichtag:** „jetzt − 12 Monate“ in UTC; archiviert wird, wer *streng
älter* ist — genau 12 Monate gilt noch nicht als „älter als“.
- **Referenzdatum:**
- Neuheiten: das Erscheinungsdatum. BGG liefert nur das Jahr, daher
großzügig das Jahresende (31.12., 23:59:59) — ein Titel von 2023 wird
erst ab dem 01.01.2025 archiviert. Ohne Jahr zählt das Eintragsdatum.
- Planungseinträge: das Eintragsdatum, unabhängig vom Status.
- **Zeitzonen:** verglichen wird durchgängig in UTC; naive Datenbank-
Zeitstempel gelten als UTC, zeitzonenbewusste Uhrzeiten werden nach UTC
verschoben (per Tests mit eingefrorener Uhr abgedeckt, inkl. 29. Februar).
### Täglicher Hintergrund-Job
Beim Start registriert das Plugin einen APScheduler-Cron-Job (Standard
täglich **03:00**, konfigurierbar über `SPIELE_ARCHIV_JOB_UHRZEIT`,
abschaltbar über `SPIELE_ARCHIV_JOB_AKTIV=0`). Pro archiviertem Titel
schreibt der Job einen **Audit-Log-Eintrag** mit dem Akteur „System“
(Aktion `verschoben`, Grund und Referenzdatum in den Details). Der
Registry-Zugriff aus dem Hintergrund-Job läuft über den neuen
`context.registry`-Hook des Plugin-Kontexts.
### Admin-Ansicht mit Wiederherstellen
Unter **Archiv** (`/archiv`, nur Rolle **Admin**) werden beide Tabellen als
vereinigte Liste angezeigt (neueste Archivierung zuerst):
- Suche über Titel/Verlag/Autor und Filter nach Herkunft (Neuheiten/Planung).
- Pro Eintrag: Herkunfts-Badge, Status, Rezensent, Archivierungszeitpunkt
und der Grund (z. B. „älter als 12 Monate (Erscheinungsjahr 2023)“).
- **Wiederherstellen** verschiebt den Eintrag vollständig zurück in die
aktive Liste und protokolliert dies im Audit-Log (Akteur: der Admin).
Konfliktfälle werden sicher abgelehnt: existiert in der Neuheitenliste
bereits ein aktiver Eintrag mit derselben BGG-ID (erneuter Sync), wird
nicht wiederhergestellt; existiert der zugeordnete Rezensent nicht mehr,
übernimmt der wiederherstellende Admin die Zuordnung (gemeldet und
auditiert). Rezensenten und Redakteure erhalten 403.
## Deployment (später)
Docker/Podman Compose ist vorgesehen (Henry-Lab, danach Kundenhardware).
@@ -379,9 +447,9 @@ geschrieben.
| 2 | Plugin `benachrichtigung` (E-Mail/Telegram/In-App, User-Präferenzen) | ✅ fertig |
| 3 | Plugin `audit-log` (Wer/Was/Wann, Admin-Ansicht, öffentliche API) | ✅ fertig |
| 4 | Plugin `neuheiten` (BGG-Sync, APScheduler, Filter) | ✅ fertig |
| 5 | Plugin `dedup` (Verlags-Konflikt, deutsche Version, Vorgänger-/Planungs-Check) | ✅ fertig (141 Tests grün) |
| 6 | Plugin `planung` (Verschiebung, händischer Eintrag + Prüfungen + Benachrichtigung) | ✅ fertig (141 Tests grün) |
| 7 | Plugin `archiv` (12-Monats-Autopilot) | ⏳ offen |
| 5 | Plugin `dedup` (Verlags-Konflikt, deutsche Version, Vorgänger-/Planungs-Check) | ✅ fertig |
| 6 | Plugin `planung` (Verschiebung, händischer Eintrag + Prüfungen + Benachrichtigung) | ✅ fertig |
| 7 | Plugin `archiv` (12-Monats-Autopilot, Wiederherstellen, täglicher Job) | ✅ fertig (171 Tests grün) |
| 8 | Plugin `erinnerung` (Redaktionsschluss pro Ausgabe, 4-Wochen-Erinnerung) | ⏳ offen |
| 9 | Plugin `export` (CSV + PDF via WeasyPrint) | ⏳ offen |
| 10 | Integrationstests über alle Plugins | ⏳ offen |

265
plugins/archiv/__init__.py
View File

@@ -1,41 +1,278 @@
"""Plugin „archiv“ — Platzhalter gemäß Plugin-Vertrag.
"""Plugin „archiv“ — automatische Archivierung nach dem Sicherungsdatei-
Prinzip.
Implementiert in einer späteren Phase. Der Stub zeigt den vollen Vertrag:
eigene Route, eigenes Template, Lifecycle-Hooks, Migrations-Schnittstelle.
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
from fastapi import Depends, Request
import logging
import os
from urllib.parse import quote
from redaktionskern.auth.deps import require_user
from redaktionskern.auth.models import User
from redaktionskern.contracts import BasePlugin, NavEntry
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)
class ArchivPlugin(BasePlugin):
name = "archiv"
title = "Archiv"
description = "Automatische Archivierung von Titeln älter als 12 Monate (Platzhalter)."
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)]
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 seite(request: Request, user: User = Depends(require_user)):
"""Platzhalterseite des Plugins."""
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,
"name": self.name,
"version": self.version,
"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,
},
)
def navigation(self) -> list[NavEntry]:
return [NavEntry(label=self.title, url="/archiv")]
@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()

108
plugins/archiv/models.py Normal file
View File

@@ -0,0 +1,108 @@
"""Datenmodelle des Plugins „archiv“.
Das Archiv folgt dem Sicherungsdatei-Prinzip: Beim Archivieren wird jede
Zeile vollständig in eine eigene Archiv-Tabelle kopiert und aus der aktiven
Liste entfernt — nichts wird verändert oder gelöscht, ein Eintrag lässt sich
jederzeit unverändert zurück in die aktive Liste verschieben.
Es gibt zwei Spiegel-Tabellen (eine pro aktiver Liste):
- `archiv_neuheiten` ← Spiegel von `neuheiten`
- `archiv_planung` ← Spiegel von `planungsliste`
Bewusst ohne Fremdschlüssel: Archivierte Zeilen müssen auch dann erhalten
bleiben, wenn z. B. der zugeordnete Rezensent gelöscht wird. Bei der
Wiederherstellung wird die Gültigkeit geprüft und notfalls ein Ersatz-
Rezensent zugeordnet.
Bewusst keine zusätzlichen Indizes: Bei doppelt geladenem Modul (direkter
Import im Test + Plugin-Loader-Namensraum) würden sie doppelt angelegt; die
Nutzerzahl ist klein.
Die Archiv-Metadaten (wer/wann/warum) stehen in jeder Zeile: `quell_id`
(ursprüngliche ID in der aktiven Liste), `archiviert_am`, `archiviert_von`,
`grund` und `referenz_am` (Zeitpunkt, an dem die 12-Monats-Regel angesetzt
hat — Erscheinungstermin bzw. Eintragsdatum).
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import DateTime, Integer, JSON, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column
from redaktionskern.db import Base
class _ArchivMeta:
"""Gemeinsame Archiv-Metadaten beider Spiegel-Tabellen (Mixin)."""
#: ID der Zeile in der ursprünglichen aktiven Liste.
quell_id: Mapped[int] = mapped_column(Integer)
#: Zeitpunkt der Archivierung (naive UTC, wie überall in der App).
archiviert_am: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
#: „System“ beim automatischen Lauf, sonst Benutzername der Wiederherstellung… # noqa
archiviert_von: Mapped[str] = mapped_column(String(100), default="System")
#: Menschlich lesbarer Grund, z. B. „älter als 12 Monate (Erscheinungsjahr 2023)“.
grund: Mapped[str] = mapped_column(String(200), default="")
#: Datum, an dem die 12-Monats-Frist angesetzt hat (Erscheinung bzw. Eintrag).
referenz_am: Mapped[datetime | None] = mapped_column(DateTime)
class ArchivNeuheit(_ArchivMeta, Base):
"""Archivierter Neuheiten-Eintrag (Spiegel der Tabelle `neuheiten`)."""
__tablename__ = "archiv_neuheiten"
# extend_existing: Das Plugin kann doppelt geladen werden (direkter Import
# im Test + Plugin-Loader-Namensraum) und teilt sich dieselbe Metadata.
__table_args__ = {"extend_existing": True}
id: Mapped[int] = mapped_column(primary_key=True)
titel: Mapped[str] = mapped_column(String(300))
verlag: Mapped[str | None] = mapped_column(String(300))
autor: Mapped[str | None] = mapped_column(String(300))
erscheinungsjahr: Mapped[int | None] = mapped_column(Integer)
bgg_id: Mapped[int | None] = mapped_column(Integer)
status: Mapped[str] = mapped_column(String(50), default="neuheit")
quelle: Mapped[str] = mapped_column(String(100), default="boardgamegeek")
erstellt_am: Mapped[datetime | None] = mapped_column(DateTime)
aktualisiert_am: Mapped[datetime | None] = mapped_column(DateTime)
def __repr__(self) -> str: # pragma: no cover - Debug-Hilfe
return f"<ArchivNeuheit {self.titel!r}>"
class ArchivPlanung(_ArchivMeta, Base):
"""Archivierter Planungseintrag (Spiegel der Tabelle `planungsliste`)."""
__tablename__ = "archiv_planung"
__table_args__ = {"extend_existing": True}
id: Mapped[int] = mapped_column(primary_key=True)
titel: Mapped[str] = mapped_column(String(300))
verlag: Mapped[str | None] = mapped_column(String(300))
autor: Mapped[str | None] = mapped_column(String(300))
bgg_id: Mapped[int | None] = mapped_column(Integer)
ausgabe: Mapped[str] = mapped_column(String(100), default="")
#: Kein FK — Archiv-Zeilen überleben das Löschen des Benutzers.
rezensent_id: Mapped[int | None] = mapped_column(Integer)
status: Mapped[str] = mapped_column(String(50), default="offen")
notizen: Mapped[str | None] = mapped_column(Text)
quelle: Mapped[str] = mapped_column(String(50), default="manuell")
pruefung: Mapped[list | None] = mapped_column(JSON)
erstellt_am: Mapped[datetime | None] = mapped_column(DateTime)
aktualisiert_am: Mapped[datetime | None] = mapped_column(DateTime)
def __repr__(self) -> str: # pragma: no cover - Debug-Hilfe
return f"<ArchivPlanung {self.titel!r}>"
#: Deutsche Anzeigenamen der Statuswerte (bewusst als Kopie ohne
#: Import-Abhängigkeit zu den Partner-Plugins, siehe planung/KONFLIKT_ANZEIGE).
STATUS_ANZEIGE: dict[str, str] = {
"neuheit": "Neuheit",
"planung": "In Planung",
"offen": "Offen",
"in_bearbeitung": "In Bearbeitung",
"abgeschlossen": "Abgeschlossen",
}

494
plugins/archiv/service.py Normal file
View File

@@ -0,0 +1,494 @@
"""Archivierungs-Logik des Plugins „archiv“.
Regeln (dokumentiert, von Tests mit eingefrorener Uhr abgedeckt):
- **Stichtag:** „jetzt minus 12 Monate“ in UTC. Archiviert wird, wer
*streng älter* als der Stichtag ist — genau 12 Monate gilt noch nicht als
„älter als 12 Monate“.
- **Referenzdatum:**
- Neuheiten: das Erscheinungsdatum. BoardGameGeek liefert nur das Jahr,
daher großzügig das Ende des Erscheinungsjahres (31.12., 23:59:59) — ein
Titel des Jahres J wird also erst nach dem 31.12. von J+1 archiviert.
Fehlt das Jahr, zählt das Eintragsdatum (`erstellt_am`).
- Planungseinträge: das Eintragsdatum (`erstellt_am`), unabhängig vom Status.
- **Zeitzonen:** Verglichen wird durchgängig in UTC. Zeitstempel ohne
Zeitzone aus der Datenbank gelten als UTC (SQLite CURRENT_TIMESTAMP ist
UTC; für Postgres werden naive Werte equally behandelt). Bewohnte
Zeitzonen werden vor dem Vergleich nach UTC verschoben.
- **Sicherungsdatei-Prinzip:** Zeilen werden vollständig in die Spiegel-
Tabellen kopiert und aus den aktiven Listen entfernt — nichts geht verloren.
"""
from __future__ import annotations
import calendar
import logging
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import datetime, timezone
from sqlalchemy import delete, insert, select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session, sessionmaker
from redaktionskern.db import Base
from .models import ArchivNeuheit, ArchivPlanung
_logger = logging.getLogger("plugins.archiv")
QUELLE_NEUHEITEN = "neuheiten"
QUELLE_PLANUNG = "planung"
GUELTIGE_QUELLEN: tuple[str, ...] = (QUELLE_NEUHEITEN, QUELLE_PLANUNG)
#: Objekttypen für Audit-Log-Einträge (deckungsgleich mit den Partner-Plugins).
OBJEKT_TYPEN: dict[str, str] = {
QUELLE_NEUHEITEN: "neuheit",
QUELLE_PLANUNG: "planungseintrag",
}
ARCHIV_ZIEL = "archiv"
GRUND_TEXT = "älter als 12 Monate"
STANDARD_MONATE = 12
def vor_monaten(moment: datetime, monate: int) -> datetime:
"""Zieht `monate` Kalendermonate ab; Monatstage werden geklemmt
(z. B. 29. Februar − 12 Monate → 28. Februar des Vorjahres)."""
monat = moment.month - monate
jahr = moment.year
while monat <= 0:
monat += 12
jahr -= 1
tag = min(moment.day, calendar.monthrange(jahr, monat)[1])
return moment.replace(year=jahr, month=monat, day=tag)
def als_utc(moment: datetime | None) -> datetime | None:
"""Naive Zeitstempel gelten als UTC; zeitzonenbewusste → UTC."""
if moment is None:
return None
if moment.tzinfo is None:
return moment.replace(tzinfo=timezone.utc)
return moment.astimezone(timezone.utc)
def naiv_utc(moment: datetime | None) -> datetime | None:
"""Nach UTC normalisiert und ohne Zeitzone (Speicherformat der App)."""
utc = als_utc(moment)
return None if utc is None else utc.replace(tzinfo=None)
@dataclass
class ArchivLauf:
"""Zusammenfassung eines automatischen Archivierungs-Laufs."""
neuheiten_archiviert: int = 0
planung_archiviert: int = 0
@property
def gesamt(self) -> int:
return self.neuheiten_archiviert + self.planung_archiviert
def als_text(self) -> str:
return (
f"{self.neuheiten_archiviert} Neuheiten und "
f"{self.planung_archiviert} Planungseinträge archiviert"
)
@dataclass
class ArchivEreignis:
"""Ein archivierter Datensatz — Grundlage für je einen Audit-Log-Eintrag."""
objekt_typ: str
objekt_id: int
details: dict = field(default_factory=dict)
@dataclass
class Wiederherstellung:
"""Ergebnis eines Wiederherstellen-Versuchs."""
ok: bool
meldung: str
ziel: str = ""
neuer_id: int | None = None
rezensent_ersetzt: bool = False
titel: str = ""
class ArchivService:
"""Verschiebt veraltete Titel in die Archiv-Tabellen und zurück."""
def __init__(
self,
session_factory: sessionmaker,
jetzt: Callable[[], datetime] | None = None,
monate: int = STANDARD_MONATE,
) -> None:
self.session_factory = session_factory
# Einfrierbare Uhr (Tests); Standard: aktuelle UTC-Zeit.
self._jetzt_fabrik = jetzt or (lambda: datetime.now(timezone.utc))
self.monate = max(int(monate), 0)
# ---------- Zeitlogik ----------
def jetzt(self) -> datetime:
return als_utc(self._jetzt_fabrik()) # type: ignore[arg-type]
def stichtag(self) -> datetime:
"""„jetzt − 12 Monate“; streng älter heißt archiviert."""
return vor_monaten(self.jetzt(), self.monate)
def referenz_neuheit(self, zeile) -> datetime:
"""Erscheinungsdatum (Ende des bekannten Jahres) oder Eintragsdatum."""
jahr = zeile["erscheinungsjahr"]
if jahr is not None:
return datetime(int(jahr), 12, 31, 23, 59, 59, tzinfo=timezone.utc)
erstellt = als_utc(zeile.get("erstellt_am"))
return erstellt or self.jetzt()
def referenz_planung(self, zeile) -> datetime:
"""Eintragsdatum des Planungseintrags."""
erstellt = als_utc(zeile.get("erstellt_am"))
return erstellt or self.jetzt()
def _grund_neuheit(self, zeile, referenz: datetime) -> str:
if zeile["erscheinungsjahr"] is not None:
return f"{GRUND_TEXT} (Erscheinungsjahr {zeile['erscheinungsjahr']})"
return f"{GRUND_TEXT} (eingetragen am {referenz:%d.%m.%Y})"
def _grund_planung(self, zeile, referenz: datetime) -> str:
return f"{GRUND_TEXT} (eingetragen am {referenz:%d.%m.%Y})"
# ---------- Automatischer Lauf ----------
def lauf(self) -> tuple[ArchivLauf, list[ArchivEreignis]]:
"""Verschiebt alle fälligen Titel in die Archiv-Tabellen.
Die Audit-Log-Einträge schreibt der Aufrufer (Route oder Hintergrund-
Job) über die öffentliche API des audit-log-Plugins — ein Ereignis
pro archiviertem Titel.
"""
stichtag = self.stichtag()
zeitpunkt = naiv_utc(self.jetzt())
lauf = ArchivLauf()
ereignisse: list[ArchivEreignis] = []
with self.session_factory() as db:
db: Session
neuheiten = Base.metadata.tables.get("neuheiten")
if neuheiten is not None:
try:
zeilen = db.execute(select(neuheiten)).mappings().all()
except SQLAlchemyError as exc:
_logger.warning(
"archiv: Tabelle 'neuheiten' nicht lesbar (%s) — "
"Quelle übersprungen.",
exc,
)
zeilen = []
for zeile in zeilen:
referenz = self.referenz_neuheit(zeile)
if referenz >= stichtag:
continue # noch keine 12 Monate alt (oder jünger)
db.execute(
insert(ArchivNeuheit).values(
quell_id=zeile["id"],
titel=zeile["titel"],
verlag=zeile["verlag"],
autor=zeile["autor"],
erscheinungsjahr=zeile["erscheinungsjahr"],
bgg_id=zeile["bgg_id"],
status=zeile["status"],
quelle=zeile["quelle"],
erstellt_am=zeile["erstellt_am"],
aktualisiert_am=zeile["aktualisiert_am"],
archiviert_am=zeitpunkt,
referenz_am=naiv_utc(referenz),
grund=self._grund_neuheit(zeile, referenz),
)
)
db.execute(
delete(neuheiten).where(neuheiten.c.id == zeile["id"])
)
lauf.neuheiten_archiviert += 1
ereignisse.append(
ArchivEreignis(
objekt_typ=OBJEKT_TYPEN[QUELLE_NEUHEITEN],
objekt_id=zeile["id"],
details={
"titel": zeile["titel"],
"herkunft": QUELLE_NEUHEITEN,
"ziel": ARCHIV_ZIEL,
"grund": self._grund_neuheit(zeile, referenz),
"referenz": f"{naiv_utc(referenz):%Y-%m-%d}",
},
)
)
planungsliste = Base.metadata.tables.get("planungsliste")
if planungsliste is not None:
try:
zeilen = db.execute(select(planungsliste)).mappings().all()
except SQLAlchemyError as exc:
_logger.warning(
"archiv: Tabelle 'planungsliste' nicht lesbar (%s) — "
"Quelle übersprungen.",
exc,
)
zeilen = []
for zeile in zeilen:
referenz = self.referenz_planung(zeile)
if referenz >= stichtag:
continue
db.execute(
insert(ArchivPlanung).values(
quell_id=zeile["id"],
titel=zeile["titel"],
verlag=zeile["verlag"],
autor=zeile["autor"],
bgg_id=zeile["bgg_id"],
ausgabe=zeile["ausgabe"],
rezensent_id=zeile["rezensent_id"],
status=zeile["status"],
notizen=zeile["notizen"],
quelle=zeile["quelle"],
pruefung=zeile["pruefung"],
erstellt_am=zeile["erstellt_am"],
aktualisiert_am=zeile["aktualisiert_am"],
archiviert_am=zeitpunkt,
referenz_am=naiv_utc(referenz),
grund=self._grund_planung(zeile, referenz),
)
)
db.execute(
delete(planungsliste).where(planungsliste.c.id == zeile["id"])
)
lauf.planung_archiviert += 1
ereignisse.append(
ArchivEreignis(
objekt_typ=OBJEKT_TYPEN[QUELLE_PLANUNG],
objekt_id=zeile["id"],
details={
"titel": zeile["titel"],
"herkunft": QUELLE_PLANUNG,
"ziel": ARCHIV_ZIEL,
"grund": self._grund_planung(zeile, referenz),
"referenz": f"{naiv_utc(referenz):%Y-%m-%d}",
},
)
)
db.commit()
if lauf.gesamt:
_logger.info("Automatische Archivierung: %s", lauf.als_text())
return lauf, ereignisse
# ---------- Ansicht ----------
def ansicht(self, db: Session) -> list[dict]:
"""Vereinigte Archiv-Ansicht beider Tabellen, neueste zuerst."""
eintraege: list[dict] = []
for a in db.scalars(
select(ArchivNeuheit).order_by(
ArchivNeuheit.archiviert_am.desc(), ArchivNeuheit.id.desc()
)
):
eintraege.append(
{
"quelle": QUELLE_NEUHEITEN,
"id": a.id,
"titel": a.titel,
"verlag": a.verlag,
"autor": a.autor,
"bgg_id": a.bgg_id,
"jahr": a.erscheinungsjahr,
"ausgabe": "",
"status": a.status,
"rezensent_id": None,
"erstellt_am": a.erstellt_am,
"archiviert_am": a.archiviert_am,
"referenz_am": a.referenz_am,
"grund": a.grund,
}
)
for a in db.scalars(
select(ArchivPlanung).order_by(
ArchivPlanung.archiviert_am.desc(), ArchivPlanung.id.desc()
)
):
eintraege.append(
{
"quelle": QUELLE_PLANUNG,
"id": a.id,
"titel": a.titel,
"verlag": a.verlag,
"autor": a.autor,
"bgg_id": a.bgg_id,
"jahr": None,
"ausgabe": a.ausgabe,
"status": a.status,
"rezensent_id": a.rezensent_id,
"erstellt_am": a.erstellt_am,
"archiviert_am": a.archiviert_am,
"referenz_am": a.referenz_am,
"grund": a.grund,
}
)
# Neueste Archivierung zuerst; bei Gleichstand die höhere ID zuerst.
return sorted(
eintraege,
key=lambda e: (e["archiviert_am"] or datetime.min, e["id"]),
reverse=True,
)
# ---------- Wiederherstellen ----------
def wiederherstellen(
self,
db: Session,
quelle: str,
archiv_id: int,
*,
ersatz_rezensent_id: int | None = None,
jetzt: datetime | None = None,
) -> Wiederherstellung:
"""Schiebt einen archivierten Eintrag zurück in die aktive Liste.
Konflikte werden abgelehnt (keine Datenänderung im Archiv): existiert
in der Neuheitenliste bereits ein aktiver Eintrag mit derselben BGG-ID,
wird nicht wiederhergestellt. Existiert der zugeordnete Rezensent
eines Planungseintrags nicht mehr, übernimmt `ersatz_rezensent_id`
die Zuordnung (wird gemeldet und auditiert).
"""
zeitpunkt = naiv_utc(jetzt or self.jetzt())
if quelle == QUELLE_NEUHEITEN:
return self._wiederherstelle_neuheit(db, archiv_id, zeitpunkt)
if quelle == QUELLE_PLANUNG:
return self._wiederherstelle_planung(
db, archiv_id, zeitpunkt, ersatz_rezensent_id
)
return Wiederherstellung(False, f"Unbekannte Herkunft „{quelle}“.")
def _wiederherstelle_neuheit(
self, db: Session, archiv_id: int, zeitpunkt: datetime
) -> Wiederherstellung:
archiv = db.get(ArchivNeuheit, archiv_id)
if archiv is None:
return Wiederherstellung(
False, "Der archivierte Eintrag wurde nicht gefunden."
)
aktiv = Base.metadata.tables.get("neuheiten")
if aktiv is None:
return Wiederherstellung(
False,
"Die Neuheitenliste ist nicht verfügbar — "
"Wiederherstellen nicht möglich.",
)
if archiv.bgg_id is not None:
vorhanden = db.execute(
select(aktiv.c.id, aktiv.c.titel).where(
aktiv.c.bgg_id == archiv.bgg_id
)
).first()
if vorhanden is not None:
return Wiederherstellung(
False,
f"Es gibt bereits einen aktiven Neuheiten-Eintrag mit der "
f"BGG-ID {archiv.bgg_id} („{vorhanden.titel}“) — "
"Wiederherstellen abgebrochen.",
)
ergebnis = db.execute(
insert(aktiv).values(
titel=archiv.titel,
verlag=archiv.verlag,
autor=archiv.autor,
erscheinungsjahr=archiv.erscheinungsjahr,
bgg_id=archiv.bgg_id,
status=archiv.status,
quelle=archiv.quelle,
erstellt_am=archiv.erstellt_am,
aktualisiert_am=zeitpunkt,
)
)
neuer_id = ergebnis.inserted_primary_key[0]
titel = archiv.titel
db.delete(archiv)
db.commit()
return Wiederherstellung(
True,
f"„{titel}“ wurde zurück in die Neuheitenliste verschoben.",
ziel=QUELLE_NEUHEITEN,
neuer_id=int(neuer_id),
titel=titel,
)
def _wiederherstelle_planung(
self,
db: Session,
archiv_id: int,
zeitpunkt: datetime,
ersatz_rezensent_id: int | None,
) -> Wiederherstellung:
archiv = db.get(ArchivPlanung, archiv_id)
if archiv is None:
return Wiederherstellung(
False, "Der archivierte Eintrag wurde nicht gefunden."
)
aktiv = Base.metadata.tables.get("planungsliste")
if aktiv is None:
return Wiederherstellung(
False,
"Die Planungsliste ist nicht verfügbar — "
"Wiederherstellen nicht möglich.",
)
rezensent_id = archiv.rezensent_id
rezensent_ersetzt = False
benutzer = Base.metadata.tables.get("users")
if rezensent_id is not None and benutzer is not None:
existiert = db.execute(
select(benutzer.c.id).where(benutzer.c.id == rezensent_id)
).first()
if existiert is None and ersatz_rezensent_id is not None:
rezensent_id = ersatz_rezensent_id
rezensent_ersetzt = True
ergebnis = db.execute(
insert(aktiv).values(
titel=archiv.titel,
verlag=archiv.verlag,
autor=archiv.autor,
bgg_id=archiv.bgg_id,
ausgabe=archiv.ausgabe,
rezensent_id=rezensent_id,
status=archiv.status,
notizen=archiv.notizen,
quelle=archiv.quelle,
pruefung=archiv.pruefung,
erstellt_am=archiv.erstellt_am,
aktualisiert_am=zeitpunkt,
)
)
neuer_id = ergebnis.inserted_primary_key[0]
titel = archiv.titel
db.delete(archiv)
db.commit()
meldung = f"„{titel}“ wurde zurück in die Planungsliste verschoben."
if rezensent_ersetzt:
meldung += (
" Der ursprüngliche Rezensent existierte nicht mehr — "
"der Eintrag wurde dir zugeordnet."
)
return Wiederherstellung(
True,
meldung,
ziel=QUELLE_PLANUNG,
neuer_id=int(neuer_id),
rezensent_ersetzt=rezensent_ersetzt,
titel=titel,
)

119
plugins/archiv/templates/archiv/index.html
View File

@@ -2,9 +2,120 @@
{% block titel %}{{ titel }} — Spiele-Redaktion{% endblock %}
{% block inhalt %}
<h1 class="text-2xl font-bold mb-2">{{ titel }}</h1>
<p class="text-slate-600 max-w-2xl">
Plugin <code class="bg-slate-200 rounded px-1 py-0.5 text-sm">{{ name }}</code>
in Version {{ version }} ist geladen.
Diese Seite ist ein Platzhalter &mdash; die Funktion wird in einer sp&auml;teren Phase implementiert.
<p class="text-slate-600 max-w-3xl mb-4">
Sicherungsdatei der Redaktion: Titel, deren Erscheinungs- bzw. Eintragsdatum
mehr als 12 Monate zur&uuml;ckliegt, werden t&auml;glich automatisch aus den
aktiven Listen hierher verschoben. Es geht nichts verloren &mdash; jeder
Eintrag l&auml;sst sich unver&auml;ndert zur&uuml;ck in die aktive Liste verschieben.
</p>
<div class="flex items-center gap-4 mb-4 text-xs text-slate-500">
<span class="inline-flex items-center gap-1.5">
<span class="w-2 h-2 rounded-full {{ 'bg-emerald-500' if job_aktiv else 'bg-slate-300' }}"></span>
T&auml;glicher Archiv-Job: {{ 'aktiv' if job_aktiv else 'aus' }}
</span>
<span>{{ anzahl_neuheiten }} archivierte Neuheiten</span>
<span>{{ anzahl_planung }} archivierte Planungseintr&auml;ge</span>
</div>
{% if meldung %}
<div class="mb-4 px-4 py-3 rounded-lg border bg-emerald-50 border-emerald-200 text-emerald-800 text-sm">{{ meldung }}</div>
{% endif %}
{% if fehler %}
<div class="mb-4 px-4 py-3 rounded-lg border bg-red-50 border-red-200 text-red-700 text-sm">{{ fehler }}</div>
{% endif %}
<div class="bg-white border border-slate-200 rounded-xl shadow-sm p-4 mb-6">
<form method="get" action="/archiv" class="flex flex-wrap items-end gap-3">
<div>
<label for="q" class="block text-xs font-medium text-slate-500 mb-1">Suche (Titel, Verlag, Autor)</label>
<input type="text" id="q" name="q" value="{{ q }}"
class="border border-slate-300 rounded-lg px-3 py-2 text-sm w-72 focus:outline-none focus:ring-2 focus:ring-emerald-500">
</div>
<div>
<label for="quelle" class="block text-xs font-medium text-slate-500 mb-1">Herkunft</label>
<select id="quelle" name="quelle"
class="border border-slate-300 rounded-lg px-3 py-2 text-sm bg-white"
onchange="this.form.submit()">
<option value="">Alle Listen</option>
<option value="neuheiten" {% if quelle_filter == 'neuheiten' %}selected{% endif %}>Neuheiten</option>
<option value="planung" {% if quelle_filter == 'planung' %}selected{% endif %}>Planung</option>
</select>
</div>
<button type="submit"
class="bg-emerald-600 hover:bg-emerald-700 text-white font-medium px-4 py-2 rounded-lg text-sm whitespace-nowrap">
Suchen
</button>
{% if q or quelle_filter %}
<a href="/archiv" class="text-sm text-slate-500 hover:text-emerald-700 px-2 py-2">Zur&uuml;cksetzen</a>
{% endif %}
</form>
</div>
<p class="text-xs text-slate-400 mb-3">{{ anzahl }} Eintr&auml;ge im Archiv</p>
<div class="overflow-x-auto bg-white rounded-xl border border-slate-200 shadow-sm">
<table class="w-full text-sm">
<thead class="bg-slate-50 text-left text-slate-500">
<tr>
<th class="px-4 py-3 font-medium">Spieltitel</th>
<th class="px-4 py-3 font-medium">Herkunft</th>
<th class="px-4 py-3 font-medium">Verlag</th>
<th class="px-4 py-3 font-medium">Jahr / Ausgabe</th>
<th class="px-4 py-3 font-medium">Status</th>
<th class="px-4 py-3 font-medium">Rezensent</th>
<th class="px-4 py-3 font-medium">Archiviert am</th>
<th class="px-4 py-3 font-medium">Grund</th>
<th class="px-4 py-3 font-medium">Aktionen</th>
</tr>
</thead>
<tbody class="divide-y divide-slate-100">
{% for eintrag in eintraege %}
<tr class="hover:bg-slate-50">
<td class="px-4 py-3 font-medium">
{{ eintrag.titel }}
{% if eintrag.bgg_id %}
<a href="https://boardgamegeek.com/boardgame/{{ eintrag.bgg_id }}" target="_blank" rel="noopener"
class="ml-1 text-[10px] text-slate-400 hover:text-emerald-700 align-super">BGG&nbsp;{{ eintrag.bgg_id }}</a>
{% endif %}
</td>
<td class="px-4 py-3 whitespace-nowrap">
{% if eintrag.quelle == 'neuheiten' %}
<span class="inline-block px-2 py-0.5 rounded-full bg-sky-100 text-sky-800 text-xs">Neuheiten</span>
{% else %}
<span class="inline-block px-2 py-0.5 rounded-full bg-violet-100 text-violet-800 text-xs">Planung</span>
{% endif %}
</td>
<td class="px-4 py-3">{{ eintrag.verlag or '—' }}</td>
<td class="px-4 py-3 whitespace-nowrap">{{ eintrag.jahr if eintrag.jahr else (eintrag.ausgabe or '—') }}</td>
<td class="px-4 py-3 whitespace-nowrap">{{ status_anzeige.get(eintrag.status, eintrag.status) }}</td>
<td class="px-4 py-3 whitespace-nowrap">{{ namen.get(eintrag.rezensent_id, '—') if eintrag.rezensent_id else '—' }}</td>
<td class="px-4 py-3 whitespace-nowrap text-slate-500">
{{ eintrag.archiviert_am.strftime('%d.%m.%Y %H:%M') if eintrag.archiviert_am else '—' }}
</td>
<td class="px-4 py-3 text-slate-500">{{ eintrag.grund }}</td>
<td class="px-4 py-3">
<form method="post" action="/archiv/{{ eintrag.quelle }}/{{ eintrag.id }}/wiederherstellen">
<button type="submit"
class="text-xs bg-emerald-600 hover:bg-emerald-700 text-white px-2.5 py-1.5 rounded-lg whitespace-nowrap"
title="Eintrag unverändert zurück in die aktive Liste verschieben">
&#8635; Wiederherstellen
</button>
</form>
</td>
</tr>
{% else %}
<tr>
<td colspan="9" class="px-4 py-8 text-center text-slate-400">
{% if q or quelle_filter %}
Keine Treffer f&uuml;r diese Suche.
{% else %}
Das Archiv ist leer — nichts zu sehen.
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% endblock %}

1
src/redaktionskern/app.py
View File

@@ -82,6 +82,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
session_factory=session_factory,
settings=settings,
templates=templates,
registry=registry,
)
with engine.begin() as conn:

6
src/redaktionskern/contracts.py
View File

@@ -22,6 +22,7 @@ if TYPE_CHECKING:
from sqlalchemy.orm import sessionmaker
from redaktionskern.config import Settings
from redaktionskern.plugin_loader import PluginRegistry
@dataclass(frozen=True)
@@ -49,11 +50,16 @@ class PluginContext:
session_factory: "sessionmaker",
settings: "Settings",
templates: "Jinja2Templates",
registry: "PluginRegistry | None" = None,
) -> None:
self.engine = engine
self.session_factory = session_factory
self.settings = settings
self.templates = templates
#: Registry aller geladenen Plugins — damit auch Hintergrund-Jobs
#: eines Plugins die öffentliche API anderer Plugins erreichen
#: (in Anfragen geht das üblicherweise über request.app.state.registry).
self.registry = registry
class BasePlugin:

2
tests/conftest.py
View File

@@ -22,6 +22,8 @@ def _hintergrundjobs_deaktiviert(monkeypatch):
monkeypatch.setenv("SPIELE_BGG_SYNC_AKTIV", "1").
"""
monkeypatch.setenv("SPIELE_BGG_SYNC_AKTIV", "0")
# Auch der tägliche Archiv-Job bleibt in Tests aus.
monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "0")
# Auch der BGG-Hilfsclient des dedup-Plugins bleibt in Tests offline.
monkeypatch.setenv("SPIELE_DEDUP_BGG_AKTIV", "0")

675
tests/test_archiv.py Normal file
View File

@@ -0,0 +1,675 @@
"""Tests: Plugin „archiv“ — 12-Monats-Regel, Job, Wiederherstellen, Rollen.
Alle Zeitpunkte werden mit eingefrorener Uhr geprüft: Der Service nimmt eine
einfrierbare „jetzt“-Fabrik entgegen (Plugin-Attribut `_jetzt` bzw.
Konstruktor-Argument), die Systemuhr wird nicht manipuliert. HTTP-Antworten
nutzen den echten TestClient; die Partner-Plugins (audit-log, neuheiten,
planung) laufen echt mit — nur ihre Tabellen werden direkt befüllt.
"""
from __future__ import annotations
import html
import tempfile
from datetime import datetime, timedelta, timezone
from fastapi.testclient import TestClient
from sqlalchemy import insert, select
from redaktionskern.app import create_app
from redaktionskern.auth.models import User
from redaktionskern.config import Settings
from redaktionskern.contracts import Migration
from redaktionskern.db import Base
from tests.conftest import ADMIN_PASSWORD, lege_benutzer_an, melde_an
from plugins.archiv.models import ArchivNeuheit, ArchivPlanung
from plugins.archiv.service import ArchivService, als_utc, naiv_utc, vor_monaten
UTC = timezone.utc
# ---------------- Hilfen ----------------
def einfriere_uhr(jahr, monat, tag, stunde=12, minute=0, sekunde=0):
"""Fabrik für eine eingefrorene UTC-Uhr."""
moment = datetime(jahr, monat, tag, stunde, minute, sekunde, tzinfo=UTC)
return lambda: moment
def archiv_service(app, jetzt) -> ArchivService:
return ArchivService(app.state.session_factory, jetzt=jetzt)
def friere_plugin_uhr_ein(app, jetzt) -> None:
app.state.registry.get("archiv")._jetzt = jetzt
def fuege_neuheit_hinzu(app, **felder) -> int:
felder.setdefault("titel", "Testspiel")
felder.setdefault("bgg_id", 4711)
tabelle = Base.metadata.tables["neuheiten"]
with app.state.session_factory() as db:
ergebnis = db.execute(insert(tabelle).values(**felder))
db.commit()
return ergebnis.inserted_primary_key[0]
def fuege_planung_hinzu(app, **felder) -> int:
felder.setdefault("titel", "Testspiel")
felder.setdefault("rezensent_id", 1)
tabelle = Base.metadata.tables["planungsliste"]
with app.state.session_factory() as db:
ergebnis = db.execute(insert(tabelle).values(**felder))
db.commit()
return ergebnis.inserted_primary_key[0]
def _alle(app, tabellenname: str) -> list[dict]:
tabelle = Base.metadata.tables[tabellenname]
with app.state.session_factory() as db:
return [dict(z) for z in db.execute(select(tabelle)).mappings().all()]
def zaehle(app, tabellenname: str) -> int:
return len(_alle(app, tabellenname))
def audit_zeilen(app) -> list[dict]:
return _alle(app, "audit_eintraege")
def sichtbar(antwort) -> str:
return html.unescape(antwort.text)
def cron_feld(trigger, name: str) -> int:
"""Liest ein CronTrigger-Feld (z. B. hour=3 → 3)."""
feld = next(f for f in trigger.fields if f.name == name)
ausdruck = feld.expressions[0]
return int(getattr(ausdruck, "first", ausdruck))
JETZT = einfriere_uhr(2025, 8, 21, 12, 0, 0) # Stichtag: 21.08.2024 12:00 UTC
# ---------------- Migration ----------------
def test_migration_legt_beide_archiv_tabellen_an(app):
from sqlalchemy import inspect
inspektion = inspect(app.state.engine)
neuheiten_spalten = {c["name"] for c in inspektion.get_columns("archiv_neuheiten")}
planung_spalten = {c["name"] for c in inspektion.get_columns("archiv_planung")}
meta = {"quell_id", "archiviert_am", "archiviert_von", "grund", "referenz_am"}
assert {
"id", "titel", "verlag", "autor", "erscheinungsjahr",
"bgg_id", "status", "quelle", "erstellt_am", "aktualisiert_am",
} | meta <= neuheiten_spalten
assert {
"id", "titel", "verlag", "autor", "bgg_id", "ausgabe", "rezensent_id",
"status", "notizen", "quelle", "pruefung",
"erstellt_am", "aktualisiert_am",
} | meta <= planung_spalten
# ---------------- Monatsarithmetik & Zeitzonen (Einheitstests) ----------------
def test_vor_monaten_klemmt_monatstage():
assert vor_monaten(datetime(2025, 3, 31), 1) == datetime(2025, 2, 28)
assert vor_monaten(datetime(2024, 2, 29), 12) == datetime(2023, 2, 28)
assert vor_monaten(datetime(2024, 3, 31), 1) == datetime(2024, 2, 29) # Schaltjahr
assert vor_monaten(datetime(2025, 1, 15), 12) == datetime(2024, 1, 15)
def test_als_utc_normalisiert_naive_und_bewohnte_zeiten():
naive = datetime(2025, 3, 10, 23, 30)
assert als_utc(naive).tzinfo == UTC
assert als_utc(naive).hour == 23 # naive Werte gelten als UTC
berlin = timezone(timedelta(hours=2))
bewohnt = datetime(2025, 3, 11, 1, 0, tzinfo=berlin) # = 23:00 UTC am 10.03.
assert als_utc(bewohnt).utcoffset() == timedelta(0)
assert (als_utc(bewohnt).month, als_utc(bewohnt).day, als_utc(bewohnt).hour) == (3, 10, 23)
assert naiv_utc(bewohnt) == datetime(2025, 3, 10, 23, 0)
def test_stichtag_rechnet_in_utc_unabhaengig_von_der_uhrzeitzone():
berlin = timezone(timedelta(hours=2))
# Die Uhr steht auf Berliner Zeit: 01:30 +02:00 == 23:30 UTC am Vortag.
service = ArchivService(
session_factory=None,
jetzt=lambda: datetime(2025, 3, 11, 1, 30, tzinfo=berlin),
)
assert service.jetzt().utcoffset() == timedelta(0)
assert (service.jetzt().hour, service.jetzt().minute) == (23, 30)
assert service.stichtag() == datetime(2024, 3, 10, 23, 30, tzinfo=UTC)
# ---------------- 12-Monats-Regel: Neuheiten ----------------
def test_alte_neuheit_wird_vollstaendig_archiviert_und_auditiert(app):
quell_id = fuege_neuheit_hinzu(
app,
titel="Alter Klassiker",
verlag="Alt-Verlag",
autor="A. Autorin",
erscheinungsjahr=2023,
bgg_id=1001,
status="neuheit",
quelle="boardgamegeek",
erstellt_am=datetime(2023, 5, 1, 10, 0),
)
lauf, ereignisse = archiv_service(app, JETZT).lauf()
# Audit-Einträge schreibt der Aufrufkontext (Route/Job) über die Plugin-API:
app.state.registry.get("archiv")._auditiere_archivierung(ereignisse)
assert (lauf.neuheiten_archiviert, lauf.planung_archiviert) == (1, 0)
assert len(ereignisse) == 1
assert zaehle(app, "neuheiten") == 0
zeile = _alle(app, "archiv_neuheiten")[0]
# Sicherungsdatei-Prinzip: alle Felder vollständig erhalten.
assert zeile["quell_id"] == quell_id
assert zeile["titel"] == "Alter Klassiker"
assert zeile["verlag"] == "Alt-Verlag"
assert zeile["autor"] == "A. Autorin"
assert zeile["erscheinungsjahr"] == 2023
assert zeile["bgg_id"] == 1001
assert zeile["status"] == "neuheit"
assert zeile["quelle"] == "boardgamegeek"
assert zeile["erstellt_am"] == datetime(2023, 5, 1, 10, 0)
# Referenz = Ende des Erscheinungsjahres (BGG liefert nur das Jahr).
assert zeile["referenz_am"] == datetime(2023, 12, 31, 23, 59, 59)
assert "2023" in zeile["grund"]
assert zeile["archiviert_von"] == "System"
# Genau ein Audit-Eintrag pro Archivierung, Akteur „System“.
eintraege = audit_zeilen(app)
assert len(eintraege) == 1
audit = eintraege[0]
assert audit["actor_name"] == "System"
assert audit["action"] == "verschoben"
assert audit["objekt_typ"] == "neuheit"
assert audit["objekt_id"] == str(quell_id)
assert audit["details"]["titel"] == "Alter Klassiker"
assert audit["details"]["ziel"] == "archiv"
assert "12 Monate" in audit["details"]["grund"]
def test_junge_neuheiten_bleiben_in_der_aktiven_liste(app):
fuege_neuheit_hinzu(app, titel="Frisch 2025", bgg_id=1, erscheinungsjahr=2025)
fuege_neuheit_hinzu(app, titel="Letztes Jahr", bgg_id=2, erscheinungsjahr=2024)
fuege_neuheit_hinzu(app, titel="Uralt", bgg_id=3, erscheinungsjahr=2023)
lauf, _ = archiv_service(app, JETZT).lauf()
# Ein Titel des Jahres 2024 zählt erst nach dem 31.12.2025 als älter als
# 12 Monate (großzügige Jahres-Auslegung) — nur 2023 fällt unter die Regel.
assert lauf.neuheiten_archiviert == 1
assert {z["titel"] for z in _alle(app, "neuheiten")} == {
"Frisch 2025", "Letztes Jahr",
}
assert {z["titel"] for z in _alle(app, "archiv_neuheiten")} == {"Uralt"}
def test_neuheit_ohne_jahr_nutzt_eintragsdatum(app):
alt = datetime(2024, 6, 1, 9, 0) # > 12 Monate vor dem 21.08.2025
jung = datetime(2024, 10, 1, 9, 0) # < 12 Monate
fuege_neuheit_hinzu(app, titel="Ohne Jahr alt", bgg_id=10,
erscheinungsjahr=None, erstellt_am=alt)
fuege_neuheit_hinzu(app, titel="Ohne Jahr jung", bgg_id=11,
erscheinungsjahr=None, erstellt_am=jung)
lauf, _ = archiv_service(app, JETZT).lauf()
assert lauf.neuheiten_archiviert == 1
zeile = _alle(app, "archiv_neuheiten")[0]
assert zeile["titel"] == "Ohne Jahr alt"
assert zeile["referenz_am"] == alt
assert "eingetragen" in zeile["grund"]
# ---------------- Grenzfälle & Planungseinträge ----------------
def test_grenzfall_genau_12_monate_bleibt_eine_sekunde_elter_faellt(app):
stichtag = datetime(2024, 3, 15, 12, 0) # exakt „jetzt − 12 Monate“
fuege_planung_hinzu(app, titel="Exakt 12 Monate", erstellt_am=stichtag)
fuege_planung_hinzu(
app, titel="Eine Sekunde älter", erstellt_am=stichtag - timedelta(seconds=1)
)
fuege_planung_hinzu(
app, titel="Eine Sekunde jünger", erstellt_am=stichtag + timedelta(seconds=1)
)
lauf, _ = archiv_service(app, einfriere_uhr(2025, 3, 15, 12, 0, 0)).lauf()
# „älter als 12 Monate“ ist streng: genau 12 Monate bleibt aktiv.
assert lauf.planung_archiviert == 1
assert {z["titel"] for z in _alle(app, "archiv_planung")} == {"Eine Sekunde älter"}
assert {z["titel"] for z in _alle(app, "planungsliste")} == {
"Exakt 12 Monate", "Eine Sekunde jünger",
}
def test_zeitzone_grenzfall_um_utc_mitternacht(app):
# Eingefrorene UTC-Uhr 23:30:30 → Stichtag 2024-03-10 23:30:30 UTC.
# In Berlin wäre es bereits der 11.03., 00:30 — die Entscheidung folgt
# trotzdem konsequent UTC.
jetzt = einfriere_uhr(2025, 3, 10, 23, 30, 30)
fuege_planung_hinzu(
app, titel="Knapp drunter", erstellt_am=datetime(2024, 3, 10, 23, 30, 0)
)
fuege_planung_hinzu(
app, titel="Knapp drueber", erstellt_am=datetime(2024, 3, 10, 23, 31, 0)
)
lauf, _ = archiv_service(app, jetzt).lauf()
assert lauf.planung_archiviert == 1
assert {z["titel"] for z in _alle(app, "archiv_planung")} == {"Knapp drunter"}
assert {z["titel"] for z in _alle(app, "planungsliste")} == {"Knapp drueber"}
def test_schaltjahr_29_februar_wird_geklemmt_nicht_uebersprungen(app):
fuege_planung_hinzu(
app, titel="29. Februar", erstellt_am=datetime(2024, 2, 29, 12, 0)
)
# 28.02.2025: Stichtag 29.02.2024 wird auf den 28.02 geklemmt — der
# Eintrag liegt einen Tag „danach“ und bleibt aktiv.
lauf, _ = archiv_service(app, einfriere_uhr(2025, 2, 28, 12, 0)).lauf()
assert lauf.planung_archiviert == 0
# 01.03.2025: Stichtag 01.03.2024 → der 29.02.2024 ist älter → Archiv.
lauf, _ = archiv_service(app, einfriere_uhr(2025, 3, 1, 12, 0)).lauf()
assert lauf.planung_archiviert == 1
assert {z["titel"] for z in _alle(app, "archiv_planung")} == {"29. Februar"}
def test_planungseintrag_wird_vollstaendig_archiviert(app):
fuege_planung_hinzu(
app,
titel="Altes Planungsspiel",
verlag="Plan-Verlag",
autor="B. Autor",
bgg_id=2002,
ausgabe="3/2024",
rezensent_id=1,
status="in_bearbeitung",
notizen="Rezension angefangen",
quelle="neuheiten",
pruefung=[{"art": "titel", "beschreibung": "ähnlicher Titel"}],
erstellt_am=datetime(2024, 5, 1, 8, 30),
)
lauf, ereignisse = archiv_service(app, JETZT).lauf()
app.state.registry.get("archiv")._auditiere_archivierung(ereignisse)
assert (lauf.neuheiten_archiviert, lauf.planung_archiviert) == (0, 1)
assert zaehle(app, "planungsliste") == 0
zeile = _alle(app, "archiv_planung")[0]
assert zeile["titel"] == "Altes Planungsspiel"
assert zeile["verlag"] == "Plan-Verlag"
assert zeile["autor"] == "B. Autor"
assert zeile["bgg_id"] == 2002
assert zeile["ausgabe"] == "3/2024"
assert zeile["status"] == "in_bearbeitung"
assert zeile["notizen"] == "Rezension angefangen"
assert zeile["quelle"] == "neuheiten"
assert zeile["pruefung"] == [{"art": "titel", "beschreibung": "ähnlicher Titel"}]
assert zeile["erstellt_am"] == datetime(2024, 5, 1, 8, 30)
assert zeile["referenz_am"] == datetime(2024, 5, 1, 8, 30)
audit = audit_zeilen(app)[0]
assert audit["objekt_typ"] == "planungseintrag"
assert audit["actor_name"] == "System"
assert audit["details"]["herkunft"] == "planung"
def test_lauf_ist_idempotent(app):
fuege_neuheit_hinzu(app, titel="Alt", bgg_id=1, erscheinungsjahr=2020)
fuege_planung_hinzu(app, titel="Alt geplant", erstellt_am=datetime(2023, 1, 1))
service = archiv_service(app, JETZT)
erster, ereignisse_eins = service.lauf()
zweiter, ereignisse_zwei = service.lauf()
plugin = app.state.registry.get("archiv")
plugin._auditiere_archivierung(ereignisse_eins)
plugin._auditiere_archivierung(ereignisse_zwei)
assert erster.gesamt == 2
assert len(ereignisse_eins) == 2
assert zweiter.gesamt == 0
assert ereignisse_zwei == []
assert zaehle(app, "archiv_neuheiten") == 1
assert zaehle(app, "archiv_planung") == 1
assert len([a for a in audit_zeilen(app) if a["actor_name"] == "System"]) == 2
def test_fehlende_partner_tabellen_brechen_nicht(tmp_path):
"""Ohne neuheiten-/planungsliste-Tabelle läuft der Lauf leer durch."""
from redaktionskern.db import make_engine, make_session_factory
engine = make_engine(f"sqlite:///{tmp_path / 'leer.db'}")
fabrik = make_session_factory(engine)
with engine.begin() as conn:
ArchivNeuheit.__table__.create(conn, checkfirst=True)
ArchivPlanung.__table__.create(conn, checkfirst=True)
try:
lauf, ereignisse = ArchivService(fabrik, jetzt=JETZT).lauf()
assert lauf.gesamt == 0
assert ereignisse == []
finally:
engine.dispose()
# ---------------- Hintergrund-Job & Scheduler ----------------
def test_job_laeuft_und_schreibt_audit_eintraege(app):
friere_plugin_uhr_ein(app, JETZT)
fuege_neuheit_hinzu(app, titel="Job-Kandidat", bgg_id=3003, erscheinungsjahr=2022)
fuege_planung_hinzu(app, titel="Job-Planung", erstellt_am=datetime(2023, 2, 2))
plugin = app.state.registry.get("archiv")
plugin._archiv_job()
assert zaehle(app, "neuheiten") == 0
assert zaehle(app, "planungsliste") == 0
assert zaehle(app, "archiv_neuheiten") == 1
assert zaehle(app, "archiv_planung") == 1
audit = audit_zeilen(app)
assert len(audit) == 2
assert {a["objekt_typ"] for a in audit} == {"neuheit", "planungseintrag"}
def test_job_ohne_audit_log_plugin_laeuft_trotzdem(app, monkeypatch):
friere_plugin_uhr_ein(app, JETZT)
fuege_neuheit_hinzu(app, titel="Ohne Audit", bgg_id=3004, erscheinungsjahr=2021)
registry = app.state.registry
original_get = registry.get
monkeypatch.setattr(
registry,
"get",
lambda name: None if name == "audit-log" else original_get(name),
)
plugin = app.state.registry.get("archiv")
plugin._archiv_job() # darf nicht werfen
assert zaehle(app, "archiv_neuheiten") == 1
assert audit_zeilen(app) == []
def test_scheduler_standardmaessig_deaktiviert_in_tests(app):
assert app.state.registry.get("archiv")._scheduler is None
def test_scheduler_laesst_sich_aktivieren_und_stoppt_beim_entladen(
monkeypatch, settings
):
monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "1")
app = create_app(settings)
plugin = app.state.registry.get("archiv")
with TestClient(app):
assert plugin._scheduler is not None
assert plugin._scheduler.running
job = plugin._scheduler.get_job("archiv-taeglich")
assert job is not None
assert cron_feld(job.trigger, "hour") == 3 # Standard 03:00 …
assert cron_feld(job.trigger, "minute") == 0 # … täglich
assert plugin._scheduler is None # on_unload hat ihn gestoppt
def test_scheduler_uhrzeit_ist_konfigurierbar(monkeypatch, settings):
monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "1")
monkeypatch.setenv("SPIELE_ARCHIV_JOB_UHRZEIT", "05:30")
app = create_app(settings)
plugin = app.state.registry.get("archiv")
with TestClient(app):
job = plugin._scheduler.get_job("archiv-taeglich")
assert cron_feld(job.trigger, "hour") == 5
assert cron_feld(job.trigger, "minute") == 30
def test_scheduler_unsinnige_uhrzeit_faellt_auf_standard(monkeypatch, settings):
monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "1")
monkeypatch.setenv("SPIELE_ARCHIV_JOB_UHRZEIT", "25:99")
app = create_app(settings)
plugin = app.state.registry.get("archiv")
with TestClient(app):
job = plugin._scheduler.get_job("archiv-taeglich")
assert cron_feld(job.trigger, "hour") == 3
assert cron_feld(job.trigger, "minute") == 0
def test_scheduler_aktiv_ist_der_standard(monkeypatch, settings):
monkeypatch.setenv("SPIELE_ARCHIV_JOB_AKTIV", "1")
app = create_app(settings)
plugin = app.state.registry.get("archiv")
with TestClient(app):
assert plugin._scheduler is not None and plugin._scheduler.running
# ---------------- Wiederherstellen ----------------
def test_wiederherstellen_neuheit_zurueck_in_aktive_liste(app, client):
melde_an(client)
friere_plugin_uhr_ein(app, JETZT)
fuege_neuheit_hinzu(
app, titel="Rückkehrer", verlag="V", autor="A",
erscheinungsjahr=2023, bgg_id=4004, status="neuheit",
erstellt_am=datetime(2023, 2, 1),
)
app.state.registry.get("archiv")._archiv_job()
archiv_id = _alle(app, "archiv_neuheiten")[0]["id"]
folgeseite = client.post(f"/archiv/neuheiten/{archiv_id}/wiederherstellen")
assert folgeseite.status_code == 200
assert zaehle(app, "archiv_neuheiten") == 0
aktiv = _alle(app, "neuheiten")
assert len(aktiv) == 1
eintrag = aktiv[0]
assert eintrag["titel"] == "Rückkehrer"
assert eintrag["bgg_id"] == 4004
assert eintrag["erscheinungsjahr"] == 2023
assert eintrag["status"] == "neuheit"
seite = sichtbar(folgeseite)
assert "zurück in die Neuheitenliste" in seite
audit = [a for a in audit_zeilen(app) if a["actor_name"] == "admin"]
assert len(audit) == 1
assert audit[0]["action"] == "verschoben"
assert audit[0]["objekt_typ"] == "neuheit"
assert audit[0]["objekt_id"] == str(eintrag["id"])
assert audit[0]["details"]["von"] == "archiv"
assert audit[0]["details"]["ziel"] == "neuheiten"
assert audit[0]["details"]["titel"] == "Rückkehrer"
def test_wiederherstellen_planung_mit_urspruenglichem_rezensenten(app, client):
lege_benutzer_an(app, "reza", "rezensent")
with app.state.session_factory() as db:
reza_id = db.scalar(select(User.id).where(User.username == "reza"))
melde_an(client)
friere_plugin_uhr_ein(app, JETZT)
fuege_planung_hinzu(
app, titel="Plan-Rückkehrer", rezensent_id=reza_id, status="offen",
ausgabe="1/2024", erstellt_am=datetime(2023, 3, 3),
)
app.state.registry.get("archiv")._archiv_job()
archiv_id = _alle(app, "archiv_planung")[0]["id"]
antwort = client.post(
f"/archiv/planung/{archiv_id}/wiederherstellen", follow_redirects=False
)
assert antwort.status_code == 303
aktiv = _alle(app, "planungsliste")
assert len(aktiv) == 1
assert aktiv[0]["rezensent_id"] == reza_id
assert aktiv[0]["status"] == "offen"
assert aktiv[0]["ausgabe"] == "1/2024"
assert zaehle(app, "archiv_planung") == 0
def test_wiederherstellen_neuheit_konflikt_bgg_id_wird_abgelehnt(app, client):
melde_an(client)
friere_plugin_uhr_ein(app, JETZT)
fuege_neuheit_hinzu(
app, titel="Konflikt-Kandidat", bgg_id=5005, erscheinungsjahr=2022
)
app.state.registry.get("archiv")._archiv_job()
# Der Sync hat den Titel inzwischen erneut geliefert — BGG-ID wieder aktiv.
fuege_neuheit_hinzu(
app, titel="Konflikt-Kandidat (neu)", bgg_id=5005, erscheinungsjahr=2025
)
archiv_id = _alle(app, "archiv_neuheiten")[0]["id"]
antwort = client.post(f"/archiv/neuheiten/{archiv_id}/wiederherstellen")
assert antwort.status_code == 200
assert "bereits einen aktiven Neuheiten-Eintrag" in sichtbar(antwort)
# Archiv unverändert, keine Duplikate entstanden.
assert zaehle(app, "archiv_neuheiten") == 1
assert zaehle(app, "neuheiten") == 1
def test_wiederherstellen_planung_ersatz_rezensent_bei_geloeschtem_benutzer(
app, client
):
lege_benutzer_an(app, "weg", "rezensent")
with app.state.session_factory() as db:
weg_id = db.scalar(select(User.id).where(User.username == "weg"))
db.delete(db.get(User, weg_id))
db.commit()
melde_an(client)
friere_plugin_uhr_ein(app, JETZT)
fuege_planung_hinzu(
app, titel="Verwaist", rezensent_id=weg_id, erstellt_am=datetime(2023, 4, 4)
)
app.state.registry.get("archiv")._archiv_job()
archiv_id = _alle(app, "archiv_planung")[0]["id"]
with app.state.session_factory() as db:
admin_id = db.scalar(select(User.id).where(User.username == "admin"))
antwort = client.post(f"/archiv/planung/{archiv_id}/wiederherstellen")
assert antwort.status_code == 200
assert "zugeordnet" in sichtbar(antwort)
aktiv = _alle(app, "planungsliste")
assert aktiv[0]["rezensent_id"] == admin_id
audit = [a for a in audit_zeilen(app) if a["actor_name"] == "admin"][0]
assert audit["details"]["rezensent_ersetzt"] is True
def test_wiederherstellen_unbekannter_eintrag_zeigt_fehler(app, client):
melde_an(client)
antwort = client.post("/archiv/neuheiten/9999/wiederherstellen")
assert antwort.status_code == 200
assert "nicht gefunden" in sichtbar(antwort)
def test_wiederherstellen_unbekannte_herkunft_wird_abgelehnt(app, client):
melde_an(client)
folgeseite = client.post("/archiv/unsinn/1/wiederherstellen")
assert folgeseite.status_code == 200
assert "Unbekannte Herkunft" in sichtbar(folgeseite)
# ---------------- Rollen ----------------
def test_archiv_ansicht_nur_fuer_admin(app, client):
# anonym → Login
antwort = client.get("/archiv", follow_redirects=False)
assert antwort.status_code == 303
assert antwort.headers["location"].startswith("/login")
# Rezensentin → 403
lege_benutzer_an(app, "rezensentin", "rezensent")
melde_an(client, "rezensentin", "test-12345678")
assert client.get("/archiv").status_code == 403
client.post("/logout")
# Redakteur → 403 (Archiv ist reine Admin-Sicht)
lege_benutzer_an(app, "redakteur", "redakteur")
melde_an(client, "redakteur", "test-12345678")
assert client.get("/archiv").status_code == 403
client.post("/logout")
# Admin → 200
melde_an(client)
assert client.get("/archiv").status_code == 200
def test_wiederherstellen_nur_fuer_admin(app, client):
friere_plugin_uhr_ein(app, JETZT)
fuege_neuheit_hinzu(app, titel="Geschützt", bgg_id=6006, erscheinungsjahr=2022)
app.state.registry.get("archiv")._archiv_job()
archiv_id = _alle(app, "archiv_neuheiten")[0]["id"]
lege_benutzer_an(app, "rezensentin", "rezensent")
melde_an(client, "rezensentin", "test-12345678")
antwort = client.post(
f"/archiv/neuheiten/{archiv_id}/wiederherstellen", follow_redirects=False
)
assert antwort.status_code == 403
assert zaehle(app, "archiv_neuheiten") == 1 # nichts verschoben
assert zaehle(app, "neuheiten") == 0
# ---------------- Admin-UI: Suche & Filter ----------------
def test_suche_filtert_ueber_titel_verlag_autor(app, client):
melde_an(client)
friere_plugin_uhr_ein(app, JETZT)
fuege_neuheit_hinzu(app, titel="Alpha Spiel", verlag="Beta Verlag",
bgg_id=1, erscheinungsjahr=2020)
fuege_neuheit_hinzu(app, titel="Gamma Spiel", autor="Delta Autorin",
bgg_id=2, erscheinungsjahr=2020)
fuege_planung_hinzu(app, titel="Epsilon Spiel",
erstellt_am=datetime(2020, 1, 1))
app.state.registry.get("archiv")._archiv_job()
def treffer(q="", quelle=""):
seite = sichtbar(client.get(f"/archiv?q={q}&quelle={quelle}"))
return [
t for t in ("Alpha Spiel", "Gamma Spiel", "Epsilon Spiel")
if t in seite
]
assert treffer(q="alpha") == ["Alpha Spiel"]
assert treffer(q="beta") == ["Alpha Spiel"] # Treffer im Verlag
assert treffer(q="delta") == ["Gamma Spiel"] # Treffer im Autor
assert treffer(quelle="planung") == ["Epsilon Spiel"]
assert treffer(quelle="neuheiten") == ["Alpha Spiel", "Gamma Spiel"]
assert treffer() == ["Alpha Spiel", "Gamma Spiel", "Epsilon Spiel"]
leer = sichtbar(client.get("/archiv?q=nirgends"))
assert "Keine Treffer" in leer
def test_archiv_seite_zeigt_details_und_job_status(app, client):
melde_an(client)
friere_plugin_uhr_ein(app, JETZT)
fuege_neuheit_hinzu(app, titel="Anzeige-Test", bgg_id=7007,
erscheinungsjahr=2021)
app.state.registry.get("archiv")._archiv_job()
seite = sichtbar(client.get("/archiv"))
assert "Anzeige-Test" in seite
assert "Erscheinungsjahr 2021" in seite # Grund der Archivierung
assert "Neuheiten" in seite # Herkunfts-Badge
assert "Wiederherstellen" in seite
assert "täglich automatisch" in seite # Erklärungstext
assert "Täglicher Archiv-Job: aus" in seite # in Tests standardmäßig aus

2
tests/test_plugin_loader.py
View File

@@ -41,7 +41,7 @@ def test_plugins_haben_navigation(app):
# Voll implementierte Plugins tragen keinen Platzhalter-Text mehr auf ihrer
# Seite; die Prüfung läuft weiter für alle übrigen Platzhalter.
VOLL_IMPLEMENTIERT = {"audit-log", "neuheiten", "benachrichtigung", "dedup", "planung"}
VOLL_IMPLEMENTIERT = {"audit-log", "neuheiten", "benachrichtigung", "dedup", "planung", "archiv"}
PLATZHALTER_STUBS = ERWARTETE_STUBS - VOLL_IMPLEMENTIERT