212 lines
8.0 KiB
Python
212 lines
8.0 KiB
Python
"""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: Titel mit Prototyp-Schlüsselwörtern werden ausgeschlossen.
|
||
- Jahresfilter: optional nur Spiele mit Erscheinungsjahr ≥ aktuelles Jahr.
|
||
- Deutsch-Filter: optional nur Spiele mit deutsch klingendem Titel.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
import re
|
||
import unicodedata
|
||
from collections.abc import Iterable, Sequence
|
||
from dataclasses import dataclass, field
|
||
from datetime import date
|
||
|
||
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
|
||
|
||
#: Funktionswörter, die stark auf einen deutschen Spieltitel hindeuten.
|
||
_DEUTSCHE_INDIKATOREN = frozenset({
|
||
"der", "die", "das", "den", "dem", "des",
|
||
"ein", "eine", "einen", "einem", "einer", "eines",
|
||
"und", "für", "von", "vom", "zu", "zum", "zur",
|
||
"im", "am", "beim", "auf", "mit", "aus",
|
||
})
|
||
#: Typische Begriffe deutscher Spieltitel (+2 Punkte je Treffer).
|
||
_DEUTSCHE_SPIELWOERTER = frozenset({
|
||
"abenteuer", "berg", "blitz", "burg", "dorf", "drache", "drachen",
|
||
"erde", "feuer", "gärten", "haus", "himmel", "insel", "jagd",
|
||
"kaiser", "könig", "könige", "königreich", "land", "länder",
|
||
"meister", "mond", "norden", "osten", "pfad", "reich", "ritter",
|
||
"schiff", "schloss", "see", "spiel", "spiele", "spielen",
|
||
"stadt", "stern", "süden", "turm", "wald", "weg", "welt",
|
||
"westen", "wiese", "wilder", "wüste", "zauber", "zeit",
|
||
})
|
||
#: Wörter, die auf einen englischen Titel hindeuten (−1 Punkt je Treffer).
|
||
_ENGLISCHE_INDIKATOREN = frozenset({
|
||
"the", "of", "and", "game", "games", "card", "cards", "dice", "board",
|
||
})
|
||
_UM_LAUTE = frozenset("äöüßÄÖÜ")
|
||
_WORT_MUSTER = re.compile(r"[^\W\d_]+", re.UNICODE)
|
||
|
||
|
||
def ist_deutscher_titel(titel: str) -> bool:
|
||
"""Heuristik: > 0 spricht für einen deutschen Titel.
|
||
|
||
Umlaute/ß zählen am stärksten (+3), deutsche Funktionswörter +1,
|
||
typische deutsche Spielbegriffe +2, englische Indikatoren −1.
|
||
"""
|
||
if not titel:
|
||
return False
|
||
score = 0
|
||
if any(zeichen in _UM_LAUTE for zeichen in titel):
|
||
score += 3
|
||
woerter = {wort.casefold() for wort in _WORT_MUSTER.findall(titel)}
|
||
score += sum(1 for wort in woerter if wort in _DEUTSCHE_INDIKATOREN)
|
||
score += sum(2 for wort in woerter if wort in _DEUTSCHE_SPIELWOERTER)
|
||
score -= sum(1 for wort in woerter if wort in _ENGLISCHE_INDIKATOREN)
|
||
return score > 0
|
||
|
||
|
||
def ist_moeglicher_prototyp(titel: str) -> bool:
|
||
"""Heuristik: Titel klingt nach Prototyp/Protoversion."""
|
||
return bool(PROTOTYP_MUSTER.search(titel))
|
||
|
||
|
||
def aktuelles_jahr() -> int:
|
||
return date.today().year
|
||
|
||
|
||
@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/Jahr/Sprache)"
|
||
)
|
||
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,
|
||
jahr_filter: bool = True,
|
||
deutscher_titel_filter: bool = True,
|
||
) -> 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
|
||
self.jahr_filter = jahr_filter
|
||
self.deutscher_titel_filter = deutscher_titel_filter
|
||
self._min_jahr = aktuelles_jahr()
|
||
|
||
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 self.jahr_filter and spiel.erscheinungsjahr is not None:
|
||
if spiel.erscheinungsjahr < self._min_jahr:
|
||
ergebnis.gefiltert += 1
|
||
continue
|
||
if self.deutscher_titel_filter and not ist_deutscher_titel(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]
|