spiele-redaktion/plugins/neuheiten/sync.py

"""Sync-Logik des Plugins „neuheiten“: BGG-Suche → Filter → Upsert.

- Update statt Duplikat: bestehende Einträge werden über die eindeutige
  BGG-ID gefunden und aktualisiert (Status bleibt erhalten).
- Erweiterungen werden im Client und hier auf Elementebene gefiltert.
- Prototypen: Die BGG-XML-API2 liefert keinen verlässlichen Prototyp-Marker;
  als Heuristik werden Titel mit Prototyp-Schlüsselwörtern ausgeschlossen
  (dokumentierte Einschränkung, siehe README).
"""
from __future__ import annotations

import logging
import re
from collections.abc import Iterable, Sequence
from dataclasses import dataclass, field

from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker

from .bgg import BggAuthFehler, BggSpiel, BggClient
from .models import STATUS_NEUHEIT, Neuheit

_logger = logging.getLogger("plugins.neuheiten")

PROTOTYP_MUSTER = re.compile(
    r"(?i)\b(prototyp\w*|prototype|proto)\b"
)

THING_BATCH_GROESSE = 20


def ist_moeglicher_prototyp(titel: str) -> bool:
    """Heuristik: Titel klingt nach Prototyp/Protoversion."""
    return bool(PROTOTYP_MUSTER.search(titel))


@dataclass
class SyncErgebnis:
    """Zusammenfassung eines Sync-Laufs."""

    neu: int = 0
    aktualisiert: int = 0
    gefiltert: int = 0
    fehler: list[str] = field(default_factory=list)
    # Grund, warum der Lauf vorzeitig endete oder ganz übersprungen wurde
    # (z. B. fehlender BGG-Token oder HTTP 401); None = normal gelaufen.
    abbruch: str | None = None

    def als_text(self) -> str:
        if self.abbruch:
            return self.abbruch
        text = (
            f"{self.neu} neu, {self.aktualisiert} aktualisiert, "
            f"{self.gefiltert} gefiltert (Erweiterungen/Prototypen)"
        )
        if self.fehler:
            text += f"; {len(self.fehler)} Fehler"
        return text


def speichere_oder_aktualisiere(db: Session, spiel: BggSpiel) -> bool:
    """Upsert nach BGG-ID. Rückgabe: True = neu angelegt, False = aktualisiert."""
    vorhanden = db.scalar(select(Neuheit).where(Neuheit.bgg_id == spiel.bgg_id))
    if vorhanden is None:
        db.add(
            Neuheit(
                titel=spiel.titel,
                verlag=spiel.verlag,
                autor=spiel.autor,
                erscheinungsjahr=spiel.erscheinungsjahr,
                bgg_id=spiel.bgg_id,
                bild_url=spiel.bild_url,
                status=STATUS_NEUHEIT,
                quelle="boardgamegeek",
            )
        )
        return True
    vorhanden.titel = spiel.titel
    vorhanden.verlag = spiel.verlag
    vorhanden.autor = spiel.autor
    vorhanden.erscheinungsjahr = spiel.erscheinungsjahr
    # Coverbild mitpflegen: BGG liefert es ggf. erst nachträglich; ein
    # fehlendes Bild in der Antwort löscht kein bereits gespeichertes.
    if spiel.bild_url:
        vorhanden.bild_url = spiel.bild_url
    return False


class SyncService:
    """Führt einen Sync-Lauf aus: Suchen → Details → Filter → Upsert."""

    def __init__(
        self,
        session_factory: sessionmaker,
        client: BggClient,
        *,
        thing_batch_groesse: int = THING_BATCH_GROESSE,
        max_treffer_pro_suchbegriff: int | None = None,
    ) -> None:
        self.session_factory = session_factory
        self.client = client
        self.thing_batch_groesse = thing_batch_groesse
        self.max_treffer_pro_suchbegriff = max_treffer_pro_suchbegriff

    def synchronisiere(self, suchbegriffe: Iterable[str]) -> SyncErgebnis:
        ergebnis = SyncErgebnis()
        with self.session_factory() as db:
            for suchbegriff in suchbegriffe:
                begriff = suchbegriff.strip()
                if not begriff:
                    continue
                try:
                    self._ein_suchbegriff(db, begriff, ergebnis)
                except BggAuthFehler as exc:
                    # HTTP 401 trifft jede weitere Anfrage genauso — der Lauf
                    # wird abgebrochen, statt alle Begriffe einzeln scheitern
                    # zu lassen; bereits gespeicherte Treffer bleiben erhalten.
                    _logger.warning("BGG-Sync abgebrochen für %r: %s", begriff, exc)
                    ergebnis.abbruch = str(exc)
                    break
                except Exception as exc:  # ein Begriff darf den Lauf nicht abbrechen
                    _logger.warning("BGG-Sync fehlgeschlagen für %r: %s", begriff, exc)
                    ergebnis.fehler.append(f"{begriff}: {exc}")
            db.commit()
        return ergebnis

    def _ein_suchbegriff(
        self, db: Session, suchbegriff: str, ergebnis: SyncErgebnis
    ) -> None:
        treffer = self.client.suche(
            suchbegriff, max_treffer=self.max_treffer_pro_suchbegriff
        )
        ids = [t.bgg_id for t in treffer]
        for batch in _batches(ids, self.thing_batch_groesse):
            for spiel in self.client.details(list(batch)):
                if spiel.typ != "boardgame":
                    ergebnis.gefiltert += 1
                    continue
                if ist_moeglicher_prototyp(spiel.titel):
                    ergebnis.gefiltert += 1
                    continue
                if speichere_oder_aktualisiere(db, spiel):
                    ergebnis.neu += 1
                else:
                    ergebnis.aktualisiert += 1


def _batches(werte: Sequence[int], groesse: int) -> Iterable[Sequence[int]]:
    for start in range(0, len(werte), groesse):
        yield werte[start : start + groesse]