96d6af8163
Teil 1 — Coverbilder: - Migration 0004_bild_url (neuheiten), 0002_bild_url (planung, archiv mit beiden Spiegel-Tabellen); nullable VARCHAR(500), SQLite↔Postgres portabel, bestehende Zeilen bleiben erhalten - BGG-Client extrahiert Cover aus der Thing-Antwort (bevorzugt <image>, sonst <thumbnail>; protokoll-relative URLs → https) - Sync-Upsert pflegt bild_url beim Update; fehlendes Bild löscht keines - Web-Quellen übernehmen img-URLs aus den Listenelementen (src/data-src, relative Pfade gegen die Seiten-URL aufgelöst); SPIEL-Essen nutzt das BILDER-Feld des JSON-Endpunkts - Verschiebung Neuheiten→Planung und Archivierung/Wiederherstellen schreiben bild_url mit - Anzeige: kleine Vorschaubilder (40x56) in den Listen, groß (max 200px) in Planung-Bearbeiten/Prüf-Dialog; Platzhalter-Div ohne Bild; Hotlinking mit loading=lazy + decoding=async Teil 2 — Responsives Layout: - Basislayout: Burger-Menü (Alpine.js) auf Mobil, Desktop-Leiste ab md: - Listen kollabieren <768px zu Karten-Ansichten (Cover links, Titel + Metadaten rechts, Status-Badge oben); Desktop behält Tabellenform - Touch-Ziele >= 44px auf Mobil, Formulare vollbreit mit Labels oben - Sync-Übersicht/Audit-Log bleiben horizontal scrollbare Tabellen Tests: 15 neue Fälle (tests/test_covers.py) — Migration inkl. Daten-Erhalt, BGG-Parsing gemockt, Quellen-Übernahme, Template-Render; 316 Tests grün. README aktualisiert.
501 lines
19 KiB
Python
501 lines
19 KiB
Python
"""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"],
|
||
bild_url=zeile["bild_url"],
|
||
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"],
|
||
bild_url=zeile["bild_url"],
|
||
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,
|
||
"bild_url": a.bild_url,
|
||
"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,
|
||
"bild_url": a.bild_url,
|
||
"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,
|
||
bild_url=archiv.bild_url,
|
||
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,
|
||
bild_url=archiv.bild_url,
|
||
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,
|
||
)
|