"""Stage 3 — human review of ambiguous/unmatched matches, then versions. A plain rich prompt loop (deliberately dependency-light, per spec). Every decision rewrites matches.csv atomically, so quitting mid-review — q, Ctrl-C, or end of input — loses nothing; the next run resumes with whatever is still ambiguous/unmatched. The version pass is optional and skippable: version review must never block getting games uploaded. """ from __future__ import annotations import json from collections.abc import Callable import httpx from rich.console import Console from rich.panel import Panel from rich.table import Table from bggpipe.bgg_client import BGGAuthError, BGGClient, BGGQueueTimeout from bggpipe.config import Config from bggpipe.models import BGGResponseError from bggpipe.resolve import ( MatchRow, _resolve_version, load_titles, read_matches, write_matches, ) # input_fn(prompt) -> user's response; tests inject a scripted one InputFn = Callable[[str], str] _BGG_ERRORS = ( BGGAuthError, BGGQueueTimeout, BGGResponseError, httpx.HTTPError, OSError, ) class _Quit(Exception): """User asked to leave the review (q / Ctrl-C / end of input).""" class ReviewSession: def __init__( self, cfg: Config, *, console: Console | None = None, input_fn: InputFn | None = None, client: BGGClient | None = None, ) -> None: self.cfg = cfg self.console = console or Console() self.input_fn = input_fn or (lambda prompt: self.console.input(prompt)) self.client = client or BGGClient(cfg.cache_dir, cfg.rate_limit_seconds) self.rows = read_matches(cfg.matches_path) self.decisions = 0 try: self.titles = load_titles(cfg.titles_path) except FileNotFoundError: self.titles = [] self._titles = {e.title_raw: e for e in self.titles} # -- plumbing ------------------------------------------------------- def _ask(self, prompt: str) -> str: try: answer = self.input_fn(prompt).strip() except (EOFError, KeyboardInterrupt) as err: raise _Quit from err if answer.lower() == "q": raise _Quit return answer def _save(self) -> None: write_matches(self.cfg.matches_path, self.rows) self.decisions += 1 def _apply_choice(self, row: dict, candidate: dict) -> None: row["match_status"] = "approved" row["bgg_id"] = str(candidate.get("bgg_id") or "") row["bgg_name"] = candidate.get("name") or "" row["year"] = str(candidate.get("year") or "") row["type"] = candidate.get("type") or "boardgame" self._fill_version(row) self._save() def _fill_version(self, row: dict) -> None: """Try version resolution for a just-approved row. Degrades gracefully: no cues, no token, or API trouble all leave version_unknown.""" entry = self._titles.get(row["title_raw"]) if entry is None or not row["bgg_id"]: row["version_status"] = row["version_status"] or "version_unknown" return shim = MatchRow(title_raw=row["title_raw"], bgg_id=int(row["bgg_id"])) try: _resolve_version(self.client, entry, shim) except _BGG_ERRORS as err: self.console.print(f"[yellow]version lookup unavailable: {err}[/yellow]") row["version_status"] = "version_unknown" return row["version_status"] = shim.version_status row["version_id"] = str(shim.version_id or "") row["version_name"] = shim.version_name row["version_candidates_json"] = json.dumps( shim.version_candidates, ensure_ascii=False ) # -- decision API (shared by the TUI and the web UI) ---------------- def pending_rows(self) -> list[dict]: return [r for r in self.rows if r["match_status"] in ("ambiguous", "unmatched")] def version_rows(self) -> list[dict]: return [ r for r in self.rows if r["version_status"] == "version_ambiguous" and r["match_status"] != "merged" ] def merged_rows(self) -> list[dict]: return [r for r in self.rows if r["match_status"] == "merged"] def veto_merge(self, row: dict) -> None: """The human says these are NOT the same physical game: restore the row as a distinct, human-confirmed match.""" row["match_status"] = "approved" row["merged_into"] = "" self._save() def cues_for(self, title_raw: str): return self._titles.get(title_raw) def decide_pick(self, row: dict, candidate: dict) -> None: self._apply_choice(row, candidate) def decide_manual(self, row: dict, bgg_id: int) -> None: self._manual_id(row, bgg_id) def decide_reject(self, row: dict) -> None: row["match_status"] = "rejected" self._save() def decide_version(self, row: dict, version_id: int | None) -> None: """Pick a version from the row's stored candidates, or None -> unknown.""" if version_id is None: row["version_status"] = "version_unknown" row["version_id"] = "" row["version_name"] = "" else: candidates = json.loads(row["version_candidates_json"] or "[]") chosen = next( (v for v in candidates if v.get("version_id") == version_id), None ) if chosen is None: raise ValueError(f"version {version_id} is not a stored candidate") row["version_status"] = "version_approved" row["version_id"] = str(version_id) row["version_name"] = chosen.get("name") or "" self._save() # -- displays ------------------------------------------------------- def _show_item(self, row: dict, candidates: list[dict]) -> None: self.console.print( Panel( f"[bold]{row['title_raw']}[/bold]\n" f"photos: {row['source_photos'] or '-'} " f"status: {row['match_status']}", expand=False, ) ) if candidates: table = Table() for col in ("#", "name", "year", "type", "owned", "rank"): table.add_column(col) for i, c in enumerate(candidates, start=1): table.add_row( str(i), str(c.get("name", "")), str(c.get("year") or "-"), str(c.get("type") or "-"), str(c.get("owned") if c.get("owned") is not None else "-"), str(c.get("rank") if c.get("rank") is not None else "-"), ) self.console.print(table) # -- match pass ----------------------------------------------------- def _review_match_row(self, row: dict) -> None: candidates = json.loads(row["candidates_json"] or "[]") while True: self._show_item(row, candidates) prompt = ( "[1-N] pick (s)kip (r)eject (m ) manual BGG id " "(f ) re-search (q)uit > " if row["match_status"] == "unmatched" else "[1-N] pick (s)kip (r)eject (q)uit > " ) answer = self._ask(prompt) lowered = answer.lower() if lowered == "s": return if lowered == "r": self.decide_reject(row) return if answer.isdigit() and 1 <= int(answer) <= len(candidates): self.decide_pick(row, candidates[int(answer) - 1]) return if lowered.startswith("m ") and answer[2:].strip().isdigit(): self._manual_id(row, int(answer[2:].strip())) return if lowered.startswith("f ") and answer[2:].strip(): candidates = self._research(answer[2:].strip()) or candidates continue self.console.print("[yellow]didn't understand that — try again[/yellow]") def _manual_id(self, row: dict, bgg_id: int) -> None: candidate: dict = {"bgg_id": bgg_id} try: (thing,) = self.client.things([bgg_id]) candidate.update(name=thing.name, year=thing.year, type=thing.type) except _BGG_ERRORS as err: self.console.print( f"[yellow]couldn't look up id {bgg_id} ({err}) — " "recording the id with no name[/yellow]" ) self._apply_choice(row, candidate) def _research(self, query: str) -> list[dict]: try: results = self.client.search(query) except _BGG_ERRORS as err: self.console.print(f"[yellow]search unavailable: {err}[/yellow]") return [] if not results: self.console.print("[yellow]no results[/yellow]") return [ {"bgg_id": r.bgg_id, "name": r.name, "year": r.year, "type": r.type} for r in results ] # -- version pass --------------------------------------------------- def _review_version_row(self, row: dict) -> None: candidates = json.loads(row["version_candidates_json"] or "[]") table = Table(title=f"{row['title_raw']} — which edition?") for col in ("#", "version", "year", "publishers", "languages", "score"): table.add_column(col) for i, v in enumerate(candidates, start=1): table.add_row( str(i), str(v.get("name", "")), str(v.get("year") or "-"), ", ".join(v.get("publishers") or []), ", ".join(v.get("languages") or []), str(v.get("score", "-")), ) self.console.print(table) while True: answer = self._ask("[1-N] pick (u)nknown (s)kip (q)uit > ") lowered = answer.lower() if lowered == "s": return if lowered == "u": self.decide_version(row, None) return if answer.isdigit() and 1 <= int(answer) <= len(candidates): self.decide_version(row, candidates[int(answer) - 1].get("version_id")) return self.console.print("[yellow]didn't understand that — try again[/yellow]") # -- entry point ---------------------------------------------------- def run(self) -> None: pending = [ r for r in self.rows if r["match_status"] in ("ambiguous", "unmatched") ] try: for row in pending: self._review_match_row(row) versions = [ r for r in self.rows if r["version_status"] == "version_ambiguous" ] if versions: answer = self._ask( f"{len(versions)} game(s) have ambiguous editions. " "Review them now? [y/N] > " ) if answer.lower() == "y": for row in versions: self._review_version_row(row) except _Quit: self.console.print("[dim]stopping — progress is saved[/dim]") remaining = sum( 1 for r in self.rows if r["match_status"] in ("ambiguous", "unmatched") ) self.console.print( f"Recorded {self.decisions} decision(s); " f"{remaining} item(s) still need review." ) def run_review( cfg: Config, *, console: Console | None = None, input_fn: InputFn | None = None, client: BGGClient | None = None, ) -> ReviewSession: session = ReviewSession(cfg, console=console, input_fn=input_fn, client=client) if not session.rows: session.console.print( f"{cfg.matches_path} is empty — run `bggpipe resolve` first." ) return session session.run() return session