spiele-redaktion/plugins/neuheiten/bgg.py

"""Client für die BoardGameGeek XML API2 (https://boardgamegeek.com/xmlapi2).

Eigenschaften:
- Rate-Limit: mindestens `mindestabstand_sekunden` (Standard 1 s) zwischen
  zwei HTTP-Requests.
- Retry mit exponentiellem Backoff bei 5xx/429; HTTP 202 (BGG-Warteschlange)
  wird gemäß Retry-After-Header erneut versucht.
- Robustes XML-Parsing: fehlende Elemente werden toleriert, kaputte
  Antworten erzeugen eine klare `BggFehler`-Ausnahme.
- Filter: Suchanfragen und Thing-Abfragen sind auf `type=boardgame`
  beschränkt; Erweiterungen (`boardgameexpansion`) werden zusätzlich auf
  Elementebene herausgefiltert.

Für Tests sind HTTP-Transport, Uhr und Schlaf-Funktion injizierbar —
es gibt keine echten Netzwerk-Aufrufe in der Testsuite.
"""
from __future__ import annotations

import time
import xml.etree.ElementTree as ET
from collections.abc import Callable, Iterable, Sequence
from dataclasses import dataclass
from typing import Any

import httpx

BASIS_URL = "https://boardgamegeek.com/xmlapi2"
USER_AGENT = "spiele-redaktion-neuheiten/0.1 (+https://github.com/local)"
TYP_BRETTSPIEL = "boardgame"
TYP_ERWEITERUNG = "boardgameexpansion"
THING_BATCH_GROESSE = 20


class BggFehler(Exception):
    """Fehler bei der Kommunikation mit oder dem Parsen der BGG-API."""


@dataclass(frozen=True)
class SuchTreffer:
    """Ein Treffer aus der BGG-Suche."""

    bgg_id: int
    titel: str
    erscheinungsjahr: int | None


@dataclass(frozen=True)
class BggSpiel:
    """Ein vollständiges Spiel aus der Thing-Abfrage."""

    bgg_id: int
    titel: str
    verlag: str | None
    autor: str | None
    erscheinungsjahr: int | None
    typ: str = TYP_BRETTSPIEL


def _attribut_wert(element: ET.Element | None, tag: str) -> str | None:
    """Liest `<tag><value>…</value></tag>` tolerant aus."""
    if element is None:
        return None
    kind = element.find(tag)
    if kind is None:
        return None
    wert = kind.get("value")
    return wert.strip() if wert and wert.strip() else None


def _jahr(wert: str | None) -> int | None:
    if not wert:
        return None
    try:
        return int(wert)
    except ValueError:
        return None


def _verketten(element: ET.Element | None, link_typ: str) -> str | None:
    """Verbindet alle `<link type="…" value="…"/>` eines Typs zu einem String."""
    if element is None:
        return None
    werte = [
        link.get("value", "").strip()
        for link in element.findall("link")
        if link.get("type") == link_typ and link.get("value", "").strip()
    ]
    return ", ".join(werte) if werte else None


def parse_suche(xml_daten: bytes | str) -> list[SuchTreffer]:
    """Parst eine Search-Antwort; nicht-Brettspiele werden verworfen."""
    try:
        wurzel = ET.fromstring(xml_daten)
    except ET.ParseError as exc:
        raise BggFehler(f"Ungültiges XML in der Suchantwort: {exc}") from exc

    treffer: list[SuchTreffer] = []
    for element in wurzel.findall("item"):
        if element.get("type") != TYP_BRETTSPIEL:
            continue  # Erweiterungen/Prototyp-Typen auf Elementebene ausschließen
        id_roh = element.get("id")
        titel = _attribut_wert(element, "name")
        if not id_roh or not titel:
            continue
        try:
            bgg_id = int(id_roh)
        except ValueError:
            continue
        treffer.append(
            SuchTreffer(
                bgg_id=bgg_id,
                titel=titel,
                erscheinungsjahr=_jahr(_attribut_wert(element, "yearpublished")),
            )
        )
    return treffer


def parse_things(xml_daten: bytes | str) -> list[BggSpiel]:
    """Parst eine Thing-Antwort; Erweiterungen werden verworfen."""
    try:
        wurzel = ET.fromstring(xml_daten)
    except ET.ParseError as exc:
        raise BggFehler(f"Ungültiges XML in der Thing-Antwort: {exc}") from exc

    spiele: list[BggSpiel] = []
    for element in wurzel.findall("item"):
        typ = element.get("type") or TYP_BRETTSPIEL
        if typ != TYP_BRETTSPIEL:
            continue  # boardgameexpansion & Co. ausschließen
        id_roh = element.get("id")
        titel = _attribut_wert(element, "name")
        if not id_roh or not titel:
            continue
        try:
            bgg_id = int(id_roh)
        except ValueError:
            continue
        spiele.append(
            BggSpiel(
                bgg_id=bgg_id,
                titel=titel,
                verlag=_verketten(element, "boardgamepublisher"),
                autor=_verketten(element, "boardgamedesigner"),
                erscheinungsjahr=_jahr(_attribut_wert(element, "yearpublished")),
                typ=typ,
            )
        )
    return spiele


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


class BggClient:
    """HTTP-Client für die XML API2 mit Rate-Limit und Retry/Backoff."""

    def __init__(
        self,
        *,
        transport: httpx.BaseTransport | None = None,
        mindestabstand_sekunden: float = 1.0,
        max_versuche: int = 4,
        backoff_basis_sekunden: float = 1.0,
        backoff_maximum_sekunden: float = 8.0,
        timeout_sekunden: float = 30.0,
        schlaf: Callable[[float], None] = time.sleep,
        uhr: Callable[[], float] = time.monotonic,
    ) -> None:
        self.mindestabstand_sekunden = mindestabstand_sekunden
        self.max_versuche = max_versuche
        self.backoff_basis_sekunden = backoff_basis_sekunden
        self.backoff_maximum_sekunden = backoff_maximum_sekunden
        self._schlaf = schlaf
        self._uhr = uhr
        self._letzter_request_um: float | None = None
        self._http = httpx.Client(
            base_url=BASIS_URL,
            timeout=timeout_sekunden,
            transport=transport,
            headers={"User-Agent": USER_AGENT},
        )

    def schliessen(self) -> None:
        self._http.close()

    def _rate_limit_abwarten(self) -> None:
        if self._letzter_request_um is None:
            return
        vergangen = self._uhr() - self._letzter_request_um
        rest = self.mindestabstand_sekunden - vergangen
        if rest > 0:
            self._schlaf(rest)

    def _hole(self, pfad: str, params: dict[str, Any]) -> bytes:
        letzte_fehler: Exception | None = None
        for versuch in range(self.max_versuche):
            self._rate_limit_abwarten()
            try:
                antwort = self._http.get(pfad, params=params)
            except httpx.HTTPError as exc:
                letzte_fehler = exc
                wartezeit = min(
                    self.backoff_basis_sekunden * (2**versuch),
                    self.backoff_maximum_sekunden,
                )
                self._schlaf(wartezeit)
                continue
            self._letzter_request_um = self._uhr()

            if antwort.status_code == 200:
                return antwort.content

            if antwort.status_code in (202, 429) or antwort.status_code >= 500:
                # 202: BGG stellt die Anfrage in eine Warteschlange;
                # 429/5xx: temporär nicht bedienbar → Backoff und erneut.
                retry_after = antwort.headers.get("Retry-After")
                if retry_after:
                    try:
                        wartezeit = float(retry_after)
                    except ValueError:
                        wartezeit = None
                else:
                    wartezeit = None
                if wartezeit is None:
                    wartezeit = min(
                        self.backoff_basis_sekunden * (2**versuch),
                        self.backoff_maximum_sekunden,
                    )
                self._schlaf(wartezeit)
                letzte_fehler = BggFehler(
                    f"HTTP {antwort.status_code} von der BGG-API (Versuch {versuch + 1})."
                )
                continue

            raise BggFehler(
                f"Unerwartete HTTP-Antwort {antwort.status_code} für {pfad}."
            )
        raise BggFehler(
            f"BGG-API nach {self.max_versuche} Versuchen nicht erreichbar"
            f" ({pfad}): {letzte_fehler}"
        )

    def suche(
        self, suchbegriff: str, *, max_treffer: int | None = None
    ) -> list[SuchTreffer]:
        """Sucht Brettspiele (`type=boardgame`) nach einem Suchbegriff."""
        treffer = parse_suche(
            self._hole("/search", {"query": suchbegriff, "type": TYP_BRETTSPIEL})
        )
        return treffer[:max_treffer] if max_treffer is not None else treffer

    def details(self, bgg_ids: Sequence[int]) -> list[BggSpiel]:
        """Lädt Thing-Details (in Batches) und filtert Erweiterungen heraus."""
        spiele: list[BggSpiel] = []
        for batch in _stuecke(list(bgg_ids), THING_BATCH_GROESSE):
            xml_daten = self._hole(
                "/thing",
                {"id": ",".join(str(i) for i in batch), "type": TYP_BRETTSPIEL},
            )
            spiele.extend(parse_things(xml_daten))
        return spiele