Resolve stage: matching, version resolution, fixtures; BGG API auth
bggpipe resolve works end to end: search -> exact/fuzzy candidate scoring -> auto/ambiguous/unmatched classification with owned-count tie-breaks (mixed base/expansion candidates never auto-match), version scoring from edition cues (never guessed; no cues -> version_unknown), idempotent matches.csv appends. Discovered mid-build: BGG now requires registered-application Bearer tokens on the XML API (2025 policy change) and returns 401 otherwise. Client sends Authorization from BGG_API_TOKEN and raises an actionable BGGAuthError; CLAUDE.md and the bgg-api skill are updated to match. Live fixture recording is blocked until registration is approved, so tests replay hand-crafted stub fixtures via a network-refusing transport; scripts/record_fixtures.py re-records real XML under the same cache keys once a token exists. One live read-only smoke test is skipped unless --run-live. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Eric Wagoner
co-authored by
Claude Fable 5
Claude Fable 5
parent
4e1211feb6
commit
2109e3544a
@@ -8,6 +8,7 @@ guarantees ≤1 request every rate_limit_seconds to any BGG endpoint.
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
import time
|
||||
@@ -36,6 +37,10 @@ class BGGQueueTimeout(Exception):
|
||||
"""BGG kept answering 202 (or throttling) past the retry budget."""
|
||||
|
||||
|
||||
class BGGAuthError(Exception):
|
||||
"""BGG rejected the request as unauthorized (missing/invalid API token)."""
|
||||
|
||||
|
||||
def cache_key(endpoint: str, params: dict[str, str]) -> str:
|
||||
query = urlencode(sorted(params.items()))
|
||||
digest = hashlib.md5(f"{endpoint}?{query}".encode()).hexdigest()[:10]
|
||||
@@ -59,10 +64,16 @@ class BGGClient:
|
||||
self._monotonic = monotonic
|
||||
self._rng = rng or random.Random()
|
||||
self._last_request: float | None = None
|
||||
headers = {"User-Agent": "bggpipe/0.1 (shelf-collection pipeline)"}
|
||||
# BGG requires registered applications since 2025: the token from
|
||||
# https://boardgamegeek.com/applications must accompany every request.
|
||||
# Env var only — never config, disk, or logs.
|
||||
if token := os.environ.get("BGG_API_TOKEN"):
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
self._http = httpx.Client(
|
||||
base_url=BASE_URL,
|
||||
timeout=30.0,
|
||||
headers={"User-Agent": "bggpipe/0.1 (shelf-collection pipeline)"},
|
||||
headers=headers,
|
||||
transport=transport,
|
||||
)
|
||||
|
||||
@@ -87,6 +98,13 @@ class BGGClient:
|
||||
if attempt < MAX_ATTEMPTS - 1:
|
||||
self._sleep(QUEUE_BACKOFF[min(attempt, len(QUEUE_BACKOFF) - 1)])
|
||||
continue
|
||||
if response.status_code == 401:
|
||||
raise BGGAuthError(
|
||||
"BGG returned 401 Unauthorized. The XML API requires a "
|
||||
"registered application token since 2025: register at "
|
||||
"https://boardgamegeek.com/applications, create a token, "
|
||||
"and export it as BGG_API_TOKEN."
|
||||
)
|
||||
if response.status_code in (429, 503):
|
||||
if attempt < MAX_ATTEMPTS - 1:
|
||||
backoff = 2.0 * (2**attempt) * (1 + self._rng.uniform(0, 0.5))
|
||||
|
||||
+346
-4
@@ -1,12 +1,354 @@
|
||||
"""Stage 2 — resolve extracted titles to BGG IDs and versions."""
|
||||
"""Stage 2 — resolve extracted titles to BGG IDs and versions.
|
||||
|
||||
Reads data/titles.json, queries BGG search (+ thing stats for tie-breaks,
|
||||
+ versions once a game is settled), classifies each title auto/ambiguous/
|
||||
unmatched, and appends rows to data/matches.csv. Re-runs skip titles
|
||||
already present in matches.csv unless --force.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
import typer
|
||||
from rapidfuzz import fuzz
|
||||
|
||||
from bggpipe.bgg_client import BGGClient
|
||||
from bggpipe.config import Config
|
||||
from bggpipe.models import GameVersion
|
||||
from bggpipe.normalize import normalize_title
|
||||
|
||||
FUZZY_THRESHOLD = 90
|
||||
# Tie-break dominance: an exact-named candidate wins outright only if it is
|
||||
# clearly the well-known game (spec: obscure duplicates lose to famous ones).
|
||||
DOMINANCE_MIN_OWNED = 100
|
||||
DOMINANCE_FACTOR = 10
|
||||
VERSION_PLAUSIBLE_SCORE = 2
|
||||
|
||||
MATCH_COLUMNS = [
|
||||
"title_raw",
|
||||
"bgg_id",
|
||||
"bgg_name",
|
||||
"year",
|
||||
"type",
|
||||
"match_status",
|
||||
"version_id",
|
||||
"version_name",
|
||||
"version_status",
|
||||
"candidates_json",
|
||||
"version_candidates_json",
|
||||
"source_photos",
|
||||
]
|
||||
|
||||
|
||||
def run_resolve(cfg: Config, *, force: bool = False) -> None:
|
||||
typer.echo("bggpipe resolve: not implemented yet (build-order step 2).")
|
||||
raise typer.Exit(code=1)
|
||||
@dataclass(frozen=True)
|
||||
class TitleEntry:
|
||||
title_raw: str
|
||||
title_normalized: str
|
||||
confidence: str = "high"
|
||||
publisher_hint: str = ""
|
||||
edition_hint: str = ""
|
||||
year_hint: int | None = None
|
||||
language_hint: str = ""
|
||||
art_notes: str = ""
|
||||
source_photos: tuple[str, ...] = ()
|
||||
|
||||
@property
|
||||
def has_version_cues(self) -> bool:
|
||||
return bool(
|
||||
self.publisher_hint
|
||||
or self.edition_hint
|
||||
or self.year_hint
|
||||
or self.language_hint
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Candidate:
|
||||
bgg_id: int
|
||||
name: str
|
||||
year: int | None
|
||||
type: str
|
||||
exact: bool
|
||||
fuzzy: float
|
||||
owned: int | None = None
|
||||
rank: int | None = None
|
||||
|
||||
def as_json(self) -> dict:
|
||||
return {
|
||||
"bgg_id": self.bgg_id,
|
||||
"name": self.name,
|
||||
"year": self.year,
|
||||
"type": self.type,
|
||||
"exact": self.exact,
|
||||
"fuzzy": round(self.fuzzy, 1),
|
||||
"owned": self.owned,
|
||||
"rank": self.rank,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class MatchRow:
|
||||
title_raw: str
|
||||
source_photos: tuple[str, ...] = ()
|
||||
bgg_id: int | None = None
|
||||
bgg_name: str = ""
|
||||
year: int | None = None
|
||||
type: str = ""
|
||||
match_status: str = "unmatched"
|
||||
version_id: int | None = None
|
||||
version_name: str = ""
|
||||
version_status: str = ""
|
||||
candidates: list[Candidate] = field(default_factory=list)
|
||||
version_candidates: list[dict] = field(default_factory=list)
|
||||
|
||||
def to_csv(self) -> dict[str, str]:
|
||||
return {
|
||||
"title_raw": self.title_raw,
|
||||
"bgg_id": str(self.bgg_id) if self.bgg_id else "",
|
||||
"bgg_name": self.bgg_name,
|
||||
"year": str(self.year) if self.year else "",
|
||||
"type": self.type,
|
||||
"match_status": self.match_status,
|
||||
"version_id": str(self.version_id) if self.version_id else "",
|
||||
"version_name": self.version_name,
|
||||
"version_status": self.version_status,
|
||||
"candidates_json": json.dumps(
|
||||
[c.as_json() for c in self.candidates], ensure_ascii=False
|
||||
),
|
||||
"version_candidates_json": json.dumps(
|
||||
self.version_candidates, ensure_ascii=False
|
||||
),
|
||||
"source_photos": ";".join(self.source_photos),
|
||||
}
|
||||
|
||||
|
||||
def load_titles(path: Path) -> list[TitleEntry]:
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(
|
||||
f"{path} not found — run `bggpipe extract` or hand-write a title list."
|
||||
)
|
||||
entries = []
|
||||
for raw in json.loads(path.read_text()):
|
||||
title_raw = raw["title_raw"]
|
||||
entries.append(
|
||||
TitleEntry(
|
||||
title_raw=title_raw,
|
||||
title_normalized=raw.get("title_normalized")
|
||||
or normalize_title(title_raw),
|
||||
confidence=raw.get("confidence", "high"),
|
||||
publisher_hint=raw.get("publisher_hint") or "",
|
||||
edition_hint=raw.get("edition_hint") or "",
|
||||
year_hint=raw.get("year_hint"),
|
||||
language_hint=raw.get("language_hint") or "",
|
||||
art_notes=raw.get("art_notes") or "",
|
||||
source_photos=tuple(raw.get("source_photos") or ()),
|
||||
)
|
||||
)
|
||||
return entries
|
||||
|
||||
|
||||
def _plausible_candidates(client: BGGClient, entry: TitleEntry) -> list[Candidate]:
|
||||
"""Search BGG and keep exact-normalized or fuzzy>=90 candidates, one per id."""
|
||||
by_id: dict[int, Candidate] = {}
|
||||
for result in client.search(entry.title_raw):
|
||||
norm = normalize_title(result.name)
|
||||
exact = norm == entry.title_normalized
|
||||
fuzzy = fuzz.token_sort_ratio(norm, entry.title_normalized)
|
||||
if not exact and fuzzy < FUZZY_THRESHOLD:
|
||||
continue
|
||||
candidate = Candidate(
|
||||
bgg_id=result.bgg_id,
|
||||
name=result.name,
|
||||
year=result.year,
|
||||
type=result.type,
|
||||
exact=exact,
|
||||
fuzzy=fuzzy,
|
||||
)
|
||||
prev = by_id.get(result.bgg_id)
|
||||
if prev is None or (candidate.exact, candidate.fuzzy) > (
|
||||
prev.exact,
|
||||
prev.fuzzy,
|
||||
):
|
||||
by_id[result.bgg_id] = candidate
|
||||
return sorted(by_id.values(), key=lambda c: (not c.exact, -c.fuzzy))
|
||||
|
||||
|
||||
def _classify(client: BGGClient, entry: TitleEntry, cands: list[Candidate]) -> MatchRow:
|
||||
row = MatchRow(title_raw=entry.title_raw, source_photos=entry.source_photos)
|
||||
if not cands:
|
||||
row.match_status = "unmatched"
|
||||
return row
|
||||
|
||||
if len(cands) == 1:
|
||||
chosen = cands[0]
|
||||
else:
|
||||
top = cands[:5]
|
||||
stats = {
|
||||
t.bgg_id: t for t in client.things([c.bgg_id for c in top], stats=True)
|
||||
}
|
||||
for c in top:
|
||||
if c.bgg_id in stats:
|
||||
c.owned = stats[c.bgg_id].owned
|
||||
c.rank = stats[c.bgg_id].rank
|
||||
row.candidates = top
|
||||
chosen = _dominant(top)
|
||||
if chosen is None:
|
||||
row.match_status = "ambiguous"
|
||||
return row
|
||||
|
||||
row.bgg_id = chosen.bgg_id
|
||||
row.bgg_name = chosen.name
|
||||
row.year = chosen.year
|
||||
row.type = chosen.type
|
||||
row.match_status = "auto"
|
||||
row.candidates = row.candidates or [chosen]
|
||||
return row
|
||||
|
||||
|
||||
def _dominant(top: list[Candidate]) -> Candidate | None:
|
||||
"""The single clear winner among plausible candidates, if any.
|
||||
|
||||
Mixed boardgame/expansion candidates are never auto-resolved (the
|
||||
spec's most common failure mode); otherwise an exact-named candidate
|
||||
wins only when its owned-count dwarfs the runner-up's.
|
||||
"""
|
||||
if len({c.type for c in top}) > 1:
|
||||
return None
|
||||
ranked = sorted(top, key=lambda c: c.owned or 0, reverse=True)
|
||||
best, second = ranked[0], ranked[1]
|
||||
if (
|
||||
best.exact
|
||||
and (best.owned or 0) >= DOMINANCE_MIN_OWNED
|
||||
and (best.owned or 0) >= DOMINANCE_FACTOR * max(second.owned or 0, 1)
|
||||
):
|
||||
return best
|
||||
return None
|
||||
|
||||
|
||||
def _score_version(entry: TitleEntry, version: GameVersion) -> int:
|
||||
score = 0
|
||||
if entry.publisher_hint:
|
||||
hint = normalize_title(entry.publisher_hint)
|
||||
if any(
|
||||
fuzz.partial_ratio(hint, normalize_title(p)) >= 85
|
||||
for p in version.publishers
|
||||
):
|
||||
score += 2
|
||||
if entry.year_hint and version.year == entry.year_hint:
|
||||
score += 2
|
||||
if entry.language_hint and entry.language_hint.casefold() in {
|
||||
lang.casefold() for lang in version.languages
|
||||
}:
|
||||
score += 1
|
||||
if (
|
||||
entry.edition_hint
|
||||
and version.name
|
||||
and fuzz.token_set_ratio(
|
||||
normalize_title(entry.edition_hint), normalize_title(version.name)
|
||||
)
|
||||
>= 80
|
||||
):
|
||||
score += 2
|
||||
return score
|
||||
|
||||
|
||||
def _resolve_version(client: BGGClient, entry: TitleEntry, row: MatchRow) -> None:
|
||||
"""Fill version_* fields on an auto/approved row. Never guess (spec)."""
|
||||
if not entry.has_version_cues:
|
||||
row.version_status = "version_unknown"
|
||||
return
|
||||
(thing,) = client.things([row.bgg_id], versions=True)
|
||||
scored = sorted(
|
||||
((v, _score_version(entry, v)) for v in thing.versions),
|
||||
key=lambda pair: -pair[1],
|
||||
)
|
||||
plausible = [(v, s) for v, s in scored if s >= VERSION_PLAUSIBLE_SCORE]
|
||||
if not plausible:
|
||||
row.version_status = "version_unknown"
|
||||
return
|
||||
row.version_candidates = [
|
||||
{
|
||||
"version_id": v.version_id,
|
||||
"name": v.name,
|
||||
"year": v.year,
|
||||
"publishers": list(v.publishers),
|
||||
"languages": list(v.languages),
|
||||
"score": s,
|
||||
}
|
||||
for v, s in plausible[:8]
|
||||
]
|
||||
if len(plausible) == 1 or plausible[0][1] > plausible[1][1]:
|
||||
winner = plausible[0][0]
|
||||
row.version_id = winner.version_id
|
||||
row.version_name = winner.name
|
||||
row.version_status = "version_auto"
|
||||
else:
|
||||
row.version_status = "version_ambiguous"
|
||||
|
||||
|
||||
def resolve_entry(client: BGGClient, entry: TitleEntry) -> MatchRow:
|
||||
row = _classify(client, entry, _plausible_candidates(client, entry))
|
||||
if row.match_status == "auto":
|
||||
_resolve_version(client, entry, row)
|
||||
return row
|
||||
|
||||
|
||||
def _row_key(title_raw: str, source_photos: str) -> tuple[str, str]:
|
||||
return (title_raw, source_photos)
|
||||
|
||||
|
||||
def read_existing_keys(path: Path) -> set[tuple[str, str]]:
|
||||
if not path.exists():
|
||||
return set()
|
||||
with path.open(newline="") as f:
|
||||
return {_row_key(r["title_raw"], r["source_photos"]) for r in csv.DictReader(f)}
|
||||
|
||||
|
||||
def append_rows(path: Path, rows: list[MatchRow]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
is_new = not path.exists()
|
||||
with path.open("a", newline="") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=MATCH_COLUMNS)
|
||||
if is_new:
|
||||
writer.writeheader()
|
||||
for row in rows:
|
||||
writer.writerow(row.to_csv())
|
||||
|
||||
|
||||
def run_resolve(
|
||||
cfg: Config, *, force: bool = False, client: BGGClient | None = None
|
||||
) -> list[MatchRow]:
|
||||
entries = load_titles(cfg.titles_path)
|
||||
if force and cfg.matches_path.exists():
|
||||
cfg.matches_path.unlink()
|
||||
existing = read_existing_keys(cfg.matches_path)
|
||||
client = client or BGGClient(cfg.cache_dir, cfg.rate_limit_seconds)
|
||||
|
||||
new_rows: list[MatchRow] = []
|
||||
skipped = 0
|
||||
for entry in entries:
|
||||
key = _row_key(entry.title_raw, ";".join(entry.source_photos))
|
||||
if key in existing:
|
||||
skipped += 1
|
||||
continue
|
||||
row = resolve_entry(client, entry)
|
||||
new_rows.append(row)
|
||||
detail = f"{row.bgg_name} ({row.bgg_id})" if row.bgg_id else "-"
|
||||
version = f" [{row.version_status}]" if row.version_status else ""
|
||||
typer.echo(f" {entry.title_raw!r} -> {row.match_status}: {detail}{version}")
|
||||
|
||||
append_rows(cfg.matches_path, new_rows)
|
||||
|
||||
counts: dict[str, int] = {}
|
||||
for row in new_rows:
|
||||
counts[row.match_status] = counts.get(row.match_status, 0) + 1
|
||||
summary = ", ".join(f"{n} {status}" for status, n in sorted(counts.items()))
|
||||
typer.echo(
|
||||
f"Resolved {len(new_rows)} title(s) ({summary or 'nothing new'}); "
|
||||
f"skipped {skipped} already in {cfg.matches_path}."
|
||||
)
|
||||
return new_rows
|
||||
|
||||
Reference in New Issue
Block a user