"""Plugin „dedup“ — Prüfungen bei Import und händischem Eintrag. Öffentliche API für andere Plugins (z. B. „planung“): dedup = request.app.state.registry.get("dedup") ergebnis = await dedup.check_titel(titel, verlag, bgg_id, user=user) # ergebnis.hat_konflikte / ergebnis.konflikte / ergebnis.verlags_optionen # ergebnis.titel_empfehlung / ergebnis.als_text() Die vier Prüfungen (Verlags-Konflikt, deutsche Titel-Variante, bereits besprochen inkl. Vorgänger-Erkennung, schon in fremder Planungsliste) sind in `pruefung.py` beschrieben. Jede Prüfung wird in der eigenen Tabelle `dedup_pruefungen` protokolliert. Zusatzdaten von BoardGameGeek (Alternate-Names, Erweiterungs-Relationen) können über SPIELE_DEDUP_BGG_AKTIV=0 abgeschaltet werden (Standard: an); ohne Netzwerk degradiert die Prüfung automatisch auf die Heuristik. """ from __future__ import annotations import logging import os from fastapi import Depends, Form, Request from sqlalchemy import select from sqlalchemy.orm import Session from redaktionskern.auth.deps import get_db, require_user from redaktionskern.auth.models import User from redaktionskern.contracts import BasePlugin, Migration, NavEntry from .bgg import BggPruefClient from .models import Pruefung from .pruefung import KONFLIKT_ANZEIGE, PruefErgebnis, fuehre_pruefung_durch _logger = logging.getLogger("plugins.dedup") ANZAHL_PROTOKOLL = 20 def _protokoll_tabelle(conn) -> None: """Migration 0001: legt die Protokoll-Tabelle an (portabel, idempotent).""" Pruefung.__table__.create(conn, checkfirst=True) class DedupPlugin(BasePlugin): name = "dedup" title = "Dedup-Prüfung" description = ( "Prüfungen bei Import und händigem Eintrag: Verlags-Konflikt, " "Titel-Varianten (deutsche Version bevorzugt), besprochene Spiele " "inkl. Vorgänger und fremde Planungseinträge." ) def __init__(self) -> None: super().__init__() self._bgg_aktiv = True self._bgg_client: BggPruefClient | None = None #: Von Tests überschreibbare Fabrik für den BGG-Hilfsclient. self.bgg_client_fabrik = None self._routen_registrieren() # ---------- Plugin-Vertrag ---------- def migrations(self) -> list[Migration]: return [Migration(version="0001_pruefprotokoll", up=_protokoll_tabelle)] def navigation(self) -> list[NavEntry]: return [NavEntry(label=self.title, url="/dedup")] def on_load(self, context) -> None: super().on_load(context) self._bgg_aktiv = os.environ.get("SPIELE_DEDUP_BGG_AKTIV", "1").strip() != "0" def on_unload(self) -> None: if self._bgg_client is not None: self._bgg_client.schliessen() self._bgg_client = None super().on_unload() # ---------- Öffentliche Prüf-API für andere Plugins ---------- async def check_titel( self, titel: str, verlag: str | None = None, bgg_id: int | str | None = None, *, user: User | None = None, ) -> PruefErgebnis: """Prüft einen Titel gegen Neuheiten- und Planungsliste. Liefert ein strukturiertes `PruefErgebnis` mit Konflikt-Details; wirft nicht (BGG-Probleme werden protokolliert und degradieren die Prüfung auf die Titel-Heuristik). Jede Prüfung landet im Protokoll. """ id_wert: int | None try: id_wert = int(bgg_id) if bgg_id not in (None, "") else None except (TypeError, ValueError): id_wert = None client = self._hilfsclient() try: ergebnis = fuehre_pruefung_durch( self.context.session_factory, client, titel=titel, verlag=verlag, bgg_id=id_wert, user_id=user.id if user is not None else None, ) finally: if client is not None: client.schliessen() self._protokolliere(ergebnis, user) return ergebnis # ---------- Internas ---------- def _hilfsclient(self) -> BggPruefClient | None: if self.bgg_client_fabrik is not None: return self.bgg_client_fabrik() if not self._bgg_aktiv or self.context is None: return None if self._bgg_client is None: self._bgg_client = BggPruefClient() return self._bgg_client def _protokolliere(self, ergebnis: PruefErgebnis, user: User | None) -> None: """Schreibt den Prüfbefund ins eigene Protokoll (best effort).""" try: with self.context.session_factory() as db: db.add( Pruefung( titel=ergebnis.titel[:300], verlag=ergebnis.verlag, bgg_id=ergebnis.bgg_id, konflikte=ergebnis.als_liste() or None, geprueft_durch_id=user.id if user is not None else None, geprueft_durch=user.username if user is not None else "System", ) ) db.commit() except Exception: _logger.exception("dedup: Prüfung konnte nicht protokolliert werden.") def _seite_rendern( self, request: Request, user: User, *, ergebnis: PruefErgebnis | None = None, meldung: str = "", fehler: str = "", ): with self.context.session_factory() as db: protokoll = db.scalars( select(Pruefung) .order_by(Pruefung.erstellt_am.desc(), Pruefung.id.desc()) .limit(ANZAHL_PROTOKOLL) ).all() return self.context.templates.TemplateResponse( request=request, name="dedup/index.html", context={ "user": user, "titel": self.title, "ergebnis": ergebnis, "konflikt_anzeige": KONFLIKT_ANZEIGE, "protokoll": protokoll, "meldung": meldung[:300], "fehler": fehler[:300], }, ) def _routen_registrieren(self) -> None: @self.router.get("/dedup") def seite(request: Request, user: User = Depends(require_user)): """Prüfseite: Manuelle Prüfung + Protokoll der letzten Prüfungen.""" return self._seite_rendern(request, user) @self.router.post("/dedup/pruefen") async def pruefen( request: Request, user: User = Depends(require_user), titel: str = Form(""), verlag: str = Form(""), bgg_id: str = Form(""), ): """Führt eine manuelle Prüfung durch und zeigt das Ergebnis.""" roh_titel = titel.strip() if not roh_titel: return self._seite_rendern( request, user, fehler="Bitte einen Spieltitel angeben." ) try: id_wert = int(bgg_id.strip()) if bgg_id.strip() else None except ValueError: return self._seite_rendern( request, user, fehler="Die BGG-ID muss eine Zahl sein.", meldung="", ) ergebnis = await self.check_titel( roh_titel, verlag.strip() or None, id_wert, user=user ) return self._seite_rendern( request, user, ergebnis=ergebnis, meldung="Prüfung abgeschlossen." if not ergebnis.hat_konflikte else "", ) plugin = DedupPlugin()