Enrich stage: full game + version metadata into games.json

Batched /thing?stats=1 fetches (20 ids, sorted so batch cache keys stay
stable), parsing the full frontend-seed payload: designers, artists,
publishers, player counts with Best-majority poll analysis, playtimes,
min age, weight, rating, rank, categories, mechanics, description, and
image URLs. Chosen-version details are reused from matches.csv's stored
candidates — zero extra API calls. Already-enriched keys are skipped
entirely; --refresh bypasses the cache read since ranks and ratings
drift. Degrades gracefully without BGG_API_TOKEN: cached ids enrich,
the rest wait, everything fetched is saved.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/bggpipe/bgg_client.py

@@ -25,6 +25,7 @@ from bggpipe.models import (
     parse_collection,
     parse_search,
     parse_things,
+    parse_things_full,
 )
 
 BASE_URL = "https://boardgamegeek.com/xmlapi2"
@@ -83,10 +84,14 @@ class BGGClient:
             if wait > 0:
                 self._sleep(wait)
 
-    def get_xml(self, endpoint: str, params: dict[str, str]) -> str:
-        """Fetch one endpoint, serving from and filling the disk cache."""
+    def get_xml(
+        self, endpoint: str, params: dict[str, str], *, refresh: bool = False
+    ) -> str:
+        """Fetch one endpoint, serving from and filling the disk cache.
+        refresh=True skips the cache read (still writes) — for data that
+        drifts over time, like ranks and ratings."""
         cache_path = self.cache_dir / cache_key(endpoint, params)
-        if cache_path.exists():
+        if cache_path.exists() and not refresh:
             return cache_path.read_text()
 
         for attempt in range(MAX_ATTEMPTS):
@@ -142,6 +147,11 @@ class BGGClient:
             params["versions"] = "1"
         return parse_things(self.get_xml("thing", params))
 
+    def things_full(self, ids: Iterable[int], *, refresh: bool = False) -> list[dict]:
+        """Full metadata dicts for the enrich stage."""
+        params = {"id": ",".join(str(i) for i in ids), "stats": "1"}
+        return parse_things_full(self.get_xml("thing", params, refresh=refresh))
+
     def collection(
         self,
         username: str,

src/bggpipe/cli.py

@@ -93,4 +93,7 @@ def enrich(
     config: ConfigOpt = None,
 ) -> None:
     """Stage 6: fetch full game + version metadata into games.json."""
-    _not_implemented("enrich", 6)
+    from bggpipe.enrich import run_enrich
+
+    cfg = load_config(config)
+    run_enrich(cfg, refresh=refresh)

src/bggpipe/enrich.py

@@ -0,0 +1,107 @@
+"""Stage 6 — enrich: full game + version metadata into games.json.
+
+The seed data for the future web frontend, keyed by bgg_id (or
+"bgg_id:version_id" when a version is settled). Cheap by design:
+- ids are batched (~20 per /thing call) and sorted so batch cache keys
+  stay stable across runs;
+- already-enriched keys are skipped entirely (no request, no cache read)
+  unless --refresh, which bypasses the cache read because ranks and
+  ratings drift over time;
+- the chosen version's details come from matches.csv's stored version
+  candidates — no extra API calls.
+Degrades gracefully without BGG_API_TOKEN: whatever is cached enriches,
+the rest waits for the token.
+"""
+
+from __future__ import annotations
+
+import json
+
+import typer
+
+from bggpipe.bgg_client import BGGAuthError, BGGClient
+from bggpipe.config import Config
+from bggpipe.resolve import read_matches
+
+BATCH_SIZE = 20
+_CONFIDENT_VERSION = ("version_auto", "version_approved")
+
+
+def _version_info(row: dict) -> dict | None:
+    if row["version_status"] not in _CONFIDENT_VERSION or not row["version_id"]:
+        return None
+    version_id = int(row["version_id"])
+    for cand in json.loads(row["version_candidates_json"] or "[]"):
+        if cand.get("version_id") == version_id:
+            return {
+                "version_id": version_id,
+                "name": cand.get("name", ""),
+                "year": cand.get("year"),
+                "publishers": cand.get("publishers") or [],
+                "languages": cand.get("languages") or [],
+            }
+    # candidates were pruned (e.g. manual review) — keep what the row knows
+    return {
+        "version_id": version_id,
+        "name": row["version_name"],
+        "year": None,
+        "publishers": [],
+        "languages": [],
+    }
+
+
+def run_enrich(
+    cfg: Config, *, refresh: bool = False, client: BGGClient | None = None
+) -> dict:
+    rows = read_matches(cfg.matches_path)
+    if not rows:
+        typer.echo(f"{cfg.matches_path} is empty — run `bggpipe resolve` first.")
+        raise typer.Exit(code=1)
+
+    targets: list[tuple[str, int, dict | None]] = []
+    for row in rows:
+        if row["match_status"] not in ("auto", "approved") or not row["bgg_id"]:
+            continue
+        version = _version_info(row)
+        key = f"{row['bgg_id']}:{version['version_id']}" if version else row["bgg_id"]
+        targets.append((key, int(row["bgg_id"]), version))
+
+    games_path = cfg.data_dir / "games.json"
+    games: dict = json.loads(games_path.read_text()) if games_path.exists() else {}
+
+    need = sorted({bgg_id for key, bgg_id, _ in targets if refresh or key not in games})
+    client = client or BGGClient(cfg.cache_dir, cfg.rate_limit_seconds)
+
+    fetched: dict[int, dict] = {}
+    blocked = False
+    for start in range(0, len(need), BATCH_SIZE):
+        batch = need[start : start + BATCH_SIZE]
+        try:
+            for thing in client.things_full(batch, refresh=refresh):
+                fetched[thing["bgg_id"]] = thing
+        except BGGAuthError:
+            blocked = True
+            break
+
+    updated = 0
+    for key, bgg_id, version in targets:
+        if bgg_id in fetched:
+            games[key] = {**fetched[bgg_id], "version": version}
+            updated += 1
+
+    games_path.parent.mkdir(parents=True, exist_ok=True)
+    games_path.write_text(json.dumps(games, indent=2, ensure_ascii=False) + "\n")
+
+    typer.echo(
+        f"games.json: {len(games)} entr{'y' if len(games) == 1 else 'ies'} "
+        f"({updated} added/refreshed this run; "
+        f"{len(targets) - updated} already present or waiting)."
+    )
+    if blocked:
+        remaining = [i for i in need if i not in fetched]
+        typer.echo(
+            f"\n{len(remaining)} game(s) are waiting on the BGG API "
+            "(set BGG_API_TOKEN and re-run enrich — everything fetched "
+            "so far is saved)."
+        )
+    return games

src/bggpipe/models.py

@@ -135,6 +135,80 @@ def parse_things(xml_text: str) -> list[ThingDetails]:
     return things
 
 
+def _attr_float(elem: ET.Element | None, attr: str = "value") -> float | None:
+    if elem is None:
+        return None
+    try:
+        return float(elem.get(attr))
+    except (TypeError, ValueError):
+        return None
+
+
+def parse_things_full(xml_text: str) -> list[dict]:
+    """Full game metadata for the enrich stage (games.json). Returns plain
+    dicts — this is artifact data, not pipeline logic."""
+    games = []
+    for item in _root(xml_text).findall("item"):
+
+        def links(link_type: str, item: ET.Element = item) -> list[str]:
+            return [
+                link.get("value", "")
+                for link in item.findall(f"link[@type='{link_type}']")
+            ]
+
+        name = item.find("name[@type='primary']")
+        ratings = item.find("statistics/ratings")
+        best_player_counts = []
+        poll = item.find("poll[@name='suggested_numplayers']")
+        if poll is not None:
+            for results in poll.findall("results"):
+                votes = {
+                    r.get("value"): int(r.get("numvotes") or 0)
+                    for r in results.findall("result")
+                }
+                best = votes.get("Best", 0)
+                if (
+                    best
+                    and best >= votes.get("Recommended", 0)
+                    and best > votes.get("Not Recommended", 0)
+                ):
+                    best_player_counts.append(results.get("numplayers"))
+        games.append(
+            {
+                "bgg_id": int(item.get("id", 0)),
+                "type": item.get("type", "boardgame"),
+                "name": name.get("value", "") if name is not None else "",
+                "year": _attr_int(item.find("yearpublished")),
+                "description": (item.findtext("description") or "").strip(),
+                "image": (item.findtext("image") or "").strip(),
+                "thumbnail": (item.findtext("thumbnail") or "").strip(),
+                "min_players": _attr_int(item.find("minplayers")),
+                "max_players": _attr_int(item.find("maxplayers")),
+                "best_player_counts": best_player_counts,
+                "playtime": _attr_int(item.find("playingtime")),
+                "min_playtime": _attr_int(item.find("minplaytime")),
+                "max_playtime": _attr_int(item.find("maxplaytime")),
+                "min_age": _attr_int(item.find("minage")),
+                "designers": links("boardgamedesigner"),
+                "artists": links("boardgameartist"),
+                "publishers": links("boardgamepublisher"),
+                "categories": links("boardgamecategory"),
+                "mechanics": links("boardgamemechanic"),
+                "rating": _attr_float(ratings.find("average"))
+                if ratings is not None
+                else None,
+                "weight": _attr_float(ratings.find("averageweight"))
+                if ratings is not None
+                else None,
+                "rank": _attr_int(item.find(".//ranks/rank[@name='boardgame']")),
+                "users_owned": _attr_int(ratings.find("owned"))
+                if ratings is not None
+                else None,
+            }
+        )
+    return games
+
+
 def parse_collection(xml_text: str) -> list[CollectionItem]:
     items = []
     for item in _root(xml_text).findall("item"):