"""Plugin „neuheiten“ — Neuheitenliste aus der BoardGameGeek XML API2. Umfang: - Eigene Migration (Tabelle `neuheiten`). - BGG-Client mit Rate-Limit/Retry (siehe bgg.py), Sync-Service (sync.py). - Regelmäßiger Sync als Hintergrund-Job (APScheduler, konfigurierbar). - Manueller „Jetzt synchronisieren“-Button für Admins/Redakteure. - Sortier-/filterbare Neuheitenliste mit Suche (deutsche UI). Konfiguration über Umgebungsvariablen (gelesen beim Plugin-Start; SPIELE_BGG_TOKEN wird dagegen vor jedem Sync gelesen): - SPIELE_BGG_SYNC_AKTIV „1“ (Standard) = Hintergrund-Job an - SPIELE_BGG_SYNC_INTERVALL_STUNDEN Intervall in Stunden (Standard 24) - SPIELE_BGG_SUCHBEGRIFFE Komma-getrennte Suchbegriffe (Standard: „brettspiel“) - SPIELE_BGG_MAX_TREFFER_PRO_SUCHE Obergrenze Treffer/Suchbegriff (25) - SPIELE_BGG_TOKEN API-Token für die BGG-XML-API2 (Pflicht seit BGG-Umstellung, wird als „Authorization: Bearer …“ gesendet); ohne Token wird der Sync übersprungen Keine Fachlogik im Kern — alles hier im Plugin gemäß Plugin-Vertrag. """ from __future__ import annotations import logging import threading from datetime import datetime, timedelta, timezone from urllib.parse import quote from fastapi import Depends, Form, Request from fastapi.responses import RedirectResponse from sqlalchemy import func, or_, select from redaktionskern.auth.deps import require_roles, require_user from redaktionskern.auth.models import Role, User from redaktionskern.contracts import BasePlugin, Migration, NavEntry, PluginContext from .bgg import BggClient, BggFehler from .models import Neuheit from .sync import SyncErgebnis, SyncService _logger = logging.getLogger("plugins.neuheiten") MELDUNG_OHNE_TOKEN = ( "Kein BGG-API-Token konfiguriert (SPIELE_BGG_TOKEN) — Sync übersprungen." ) SORTIERBAR = { "titel": Neuheit.titel, "verlag": Neuheit.verlag, "autor": Neuheit.autor, "jahr": Neuheit.erscheinungsjahr, "status": Neuheit.status, "aktualisiert": Neuheit.aktualisiert_am, } STANDARD_SORTIERUNG = ("titel", "auf") MAX_MELDUNGS_LAENGE = 300 def _umgebung_liste(name: str, standard: str) -> list[str]: import os roh = os.environ.get(name, standard) return [teil.strip() for teil in roh.split(",") if teil.strip()] class NeuheitenPlugin(BasePlugin): name = "neuheiten" title = "Neuheiten" description = ( "Neuheitenliste aus der BoardGameGeek-API — regelmäßiger Sync als " "Hintergrund-Job, Erweiterungen und Prototypen werden gefiltert." ) def __init__(self) -> None: super().__init__() self._scheduler = None # BackgroundScheduler, falls aktiviert self._sync_sperre = threading.Lock() self._suchbegriffe: list[str] = [] self._max_treffer_pro_suche = 25 @self.router.get("/neuheiten") def seite( request: Request, user: User = Depends(require_user), q: str = "", status: str = "", sort: str = STANDARD_SORTIERUNG[0], richtung: str = STANDARD_SORTIERUNG[1], meldung: str = "", fehler: str = "", ): spalte = SORTIERBAR.get(sort, SORTIERBAR[STANDARD_SORTIERUNG[0]]) abwaerts = richtung == "ab" if sort in SORTIERBAR else False spalte_sortiert = spalte.desc() if abwaerts else spalte.asc() with self.context.session_factory() as db: abfrage = select(Neuheit) if q.strip(): muster = f"%{q.strip().lower()}%" abfrage = abfrage.where( or_( func.lower(Neuheit.titel).like(muster), func.lower(func.coalesce(Neuheit.verlag, "")).like(muster), func.lower(func.coalesce(Neuheit.autor, "")).like(muster), ) ) if status.strip(): abfrage = abfrage.where(Neuheit.status == status.strip()) eintraege = db.scalars( abfrage.order_by(spalte_sortiert, Neuheit.id.asc()) ).all() status_optionen = [ zeile for zeile in db.scalars( select(Neuheit.status).distinct().order_by(Neuheit.status) ) ] kontext = { "user": user, "titel": self.title, "eintraege": eintraege, "q": q, "status_filter": status, "sort": sort if sort in SORTIERBAR else STANDARD_SORTIERUNG[0], "richtung": "ab" if abwaerts else "auf", "status_optionen": status_optionen, "meldung": meldung[:MAX_MELDUNGS_LAENGE], "fehler": fehler[:MAX_MELDUNGS_LAENGE], "ist_redaktion": user.role in ( Role.ADMIN.value, Role.REDAKTEUR.value, ), "suchbegriffe": ", ".join(self._suchbegriffe) or "—", "sync_aktiv": self._scheduler is not None and self._scheduler.running, "anzahl": len(eintraege), } if request.headers.get("HX-Request") == "true": return self.context.templates.TemplateResponse( request=request, name="neuheiten/_liste.html", context=kontext, ) return self.context.templates.TemplateResponse( request=request, name="neuheiten/index.html", context=kontext, ) @self.router.post("/neuheiten/sync") def jetzt_synchronisieren( user: User = Depends(require_roles(Role.ADMIN.value, Role.REDAKTEUR.value)), suchbegriff: str = Form(""), ): begriffe = [suchbegriff] if suchbegriff.strip() else self._suchbegriffe try: ergebnis = self._sync_ausfuehren(begriffe) except BggFehler as exc: _logger.warning("Manueller BGG-Sync fehlgeschlagen: %s", exc) ziel = f"/neuheiten?meldung={quote(f'Sync fehlgeschlagen: {exc}')}" return RedirectResponse(ziel, status_code=303) if ergebnis.abbruch: # z. B. fehlender Token oder HTTP 401 — klar benennbare Ursache ziel = f"/neuheiten?fehler={quote(ergebnis.abbruch)}" else: ziel = f"/neuheiten?meldung={quote(f'Sync abgeschlossen: {ergebnis.als_text()}')}" return RedirectResponse(ziel, status_code=303) # ---------- Plugin-Vertrag ---------- def migrations(self) -> list[Migration]: def neuheiten_tabelle(conn) -> None: Neuheit.__table__.create(conn, checkfirst=True) return [Migration(version="0001_neuheiten_tabelle", up=neuheiten_tabelle)] def navigation(self) -> list[NavEntry]: return [NavEntry(label=self.title, url="/neuheiten")] def on_load(self, context: PluginContext) -> None: super().on_load(context) import os self._suchbegriffe = _umgebung_liste("SPIELE_BGG_SUCHBEGRIFFE", "brettspiel") try: self._max_treffer_pro_suche = int( os.environ.get("SPIELE_BGG_MAX_TREFFER_PRO_SUCHE", "25") ) except ValueError: self._max_treffer_pro_suche = 25 if os.environ.get("SPIELE_BGG_SYNC_AKTIV", "1").strip() == "1": self._scheduler_starten(os.environ.get("SPIELE_BGG_SYNC_INTERVALL_STUNDEN")) def on_unload(self) -> None: if self._scheduler is not None: self._scheduler.shutdown(wait=False) self._scheduler = None super().on_unload() # ---------- Internas ---------- def _aktives_bgg_token(self) -> str | None: """Liest den BGG-API-Token aus der Umgebung (leer = nicht gesetzt). Bewusst pro Sync gelesen, damit ein rotierter Token ohne Neustart greift. """ import os return (os.environ.get("SPIELE_BGG_TOKEN") or "").strip() or None def _neuer_client(self) -> BggClient: """Fabrik für den BGG-Client; von Tests überschreibbar.""" return BggClient(token=self._aktives_bgg_token()) def _sync_ausfuehren(self, suchbegriffe: list[str]): if self._aktives_bgg_token() is None: # Ohne Token lehnt die BGG-API jeden Request mit 401 ab — # den Sync gar nicht erst starten. _logger.warning(MELDUNG_OHNE_TOKEN) return SyncErgebnis(abbruch=MELDUNG_OHNE_TOKEN) client = self._neuer_client() try: service = SyncService( self.context.session_factory, client, max_treffer_pro_suchbegriff=self._max_treffer_pro_suche, ) return service.synchronisiere(suchbegriffe) finally: client.schliessen() def _sync_job(self) -> None: """Hintergrund-Job: holt regelmäßig die Neuheiten ab.""" if not self._sync_sperre.acquire(blocking=False): _logger.info("BGG-Sync läuft bereits — Durchlauf übersprungen.") return try: ergebnis = self._sync_ausfuehren(self._suchbegriffe) if ergebnis.abbruch: _logger.warning("BGG-Sync nicht ausgeführt: %s", ergebnis.abbruch) else: _logger.info("BGG-Sync abgeschlossen: %s", ergebnis.als_text()) except Exception as exc: _logger.warning("BGG-Sync fehlgeschlagen: %s", exc) finally: self._sync_sperre.release() def _scheduler_starten(self, intervall_stunden_roh: str | None) -> None: from apscheduler.schedulers.background import BackgroundScheduler try: intervall_stunden = int(intervall_stunden_roh or "24") except ValueError: intervall_stunden = 24 intervall_stunden = max(intervall_stunden, 1) scheduler = BackgroundScheduler() erster_lauf = datetime.now(timezone.utc) + timedelta(seconds=10) scheduler.add_job( self._sync_job, trigger="interval", hours=intervall_stunden, next_run_time=erster_lauf, id="bgg-neuheiten-sync", replace_existing=True, ) scheduler.start() self._scheduler = scheduler _logger.info( "BGG-Neuheiten-Sync aktiv: alle %s h, Suchbegriffe: %s", intervall_stunden, ", ".join(self._suchbegriffe), ) plugin = NeuheitenPlugin()