Review stage: resumable rich TUI for matches and versions

Prompt loop over ambiguous/unmatched rows: pick a candidate (table with
owned/rank), skip, reject, enter a manual BGG id, or free-text re-search
via the cached client. Approvals attempt version resolution from the
title's edition cues, degrading to version_unknown when the API is
unreachable (no token yet). Optional, skippable version pass for
version_ambiguous rows. Every decision rewrites matches.csv atomically,
so q/Ctrl-C/EOF mid-session loses nothing. Tests drive the loop with
scripted input against a synthetic matches.csv and the fixture cache.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Eric Wagoner
2026-08-01 13:13:20 -04:00
parent b7a1ef8549
commit 9897da7f82
4 changed files with 590 additions and 1 deletions
+4 -1
View File
@@ -58,7 +58,10 @@ def resolve(
@app.command()
def review(config: ConfigOpt = None) -> None:
"""Stage 3: human review of ambiguous/unmatched items."""
_not_implemented("review", 4)
from bggpipe.review import run_review
cfg = load_config(config)
run_review(cfg)
@app.command()
+19
View File
@@ -10,6 +10,7 @@ from __future__ import annotations
import csv
import json
import os
from dataclasses import dataclass, field
from pathlib import Path
@@ -301,6 +302,24 @@ def _row_key(title_raw: str, source_photos: str) -> tuple[str, str]:
return (title_raw, source_photos)
def read_matches(path: Path) -> list[dict[str, str]]:
if not path.exists():
return []
with path.open(newline="") as f:
return list(csv.DictReader(f))
def write_matches(path: Path, rows: list[dict[str, str]]) -> None:
"""Atomic full rewrite — review updates rows in place decision by decision."""
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_name(path.name + ".tmp")
with tmp.open("w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=MATCH_COLUMNS, extrasaction="ignore")
writer.writeheader()
writer.writerows(rows)
os.replace(tmp, path)
def read_existing_keys(path: Path) -> set[tuple[str, str]]:
if not path.exists():
return set()
+277
View File
@@ -0,0 +1,277 @@
"""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 = {e.title_raw: e for e in load_titles(cfg.titles_path)}
except FileNotFoundError:
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
)
# -- 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 <id>) manual BGG id "
"(f <text>) 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":
row["match_status"] = "rejected"
self._save()
return
if answer.isdigit() and 1 <= int(answer) <= len(candidates):
self._apply_choice(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":
row["version_status"] = "version_unknown"
row["version_id"] = ""
row["version_name"] = ""
self._save()
return
if answer.isdigit() and 1 <= int(answer) <= len(candidates):
chosen = candidates[int(answer) - 1]
row["version_status"] = "version_approved"
row["version_id"] = str(chosen.get("version_id") or "")
row["version_name"] = chosen.get("name") or ""
self._save()
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
+290
View File
@@ -0,0 +1,290 @@
"""Review-TUI tests against a synthetic matches.csv, driven by scripted
input. BGG responses replay from the fixture cache; nothing hits the
network (the transport raises if it would)."""
from __future__ import annotations
import io
import json
from pathlib import Path
import httpx
from rich.console import Console
from bggpipe.bgg_client import BGGClient
from bggpipe.config import Config
from bggpipe.resolve import read_matches, write_matches
from bggpipe.review import run_review
FIXTURES = Path(__file__).parent / "fixtures" / "bgg_cache"
def _no_network(request: httpx.Request) -> httpx.Response:
raise AssertionError(f"test hit the network: {request.url}")
def fixture_client() -> BGGClient:
return BGGClient(cache_dir=FIXTURES, transport=httpx.MockTransport(_no_network))
def unauthorized_client(tmp_path) -> BGGClient:
"""Client whose every request 401s — the no-token-yet world."""
transport = httpx.MockTransport(
lambda req: httpx.Response(401, text="Unauthorized")
)
return BGGClient(cache_dir=tmp_path / "empty_cache", transport=transport)
def scripted(*answers):
it = iter(answers)
def input_fn(prompt: str) -> str:
try:
return next(it)
except StopIteration:
raise EOFError from None
return input_fn
def quiet_console() -> Console:
return Console(file=io.StringIO(), width=100)
def _row(**overrides) -> dict:
row = {
"title_raw": "",
"bgg_id": "",
"bgg_name": "",
"year": "",
"type": "",
"match_status": "auto",
"version_id": "",
"version_name": "",
"version_status": "version_unknown",
"candidates_json": "[]",
"version_candidates_json": "[]",
"source_photos": "hand-typed-test-list",
}
row.update(overrides)
return row
CITADELS_CANDIDATES = json.dumps(
[
{
"bgg_id": 478,
"name": "Citadels",
"year": 2000,
"type": "boardgame",
"owned": 85000,
"rank": 250,
},
{
"bgg_id": 205398,
"name": "Citadels",
"year": 2016,
"type": "boardgame",
"owned": 24000,
"rank": 400,
},
]
)
WINGSPAN_VERSION_CANDIDATES = json.dumps(
[
{
"version_id": 465063,
"name": "English first edition",
"year": 2019,
"publishers": ["Stonemaier Games"],
"languages": ["English"],
"score": 5,
},
{
"version_id": 521212,
"name": "English fourth printing",
"year": 2020,
"publishers": ["Stonemaier Games"],
"languages": ["English"],
"score": 5,
},
]
)
def _cfg(tmp_path) -> Config:
return Config(data_dir=tmp_path / "data")
def _setup(tmp_path, rows) -> Config:
cfg = _cfg(tmp_path)
write_matches(cfg.matches_path, rows)
return cfg
def test_ambiguous_pick_approves_candidate(tmp_path):
cfg = _setup(
tmp_path,
[
_row(
title_raw="Citadels",
match_status="ambiguous",
candidates_json=CITADELS_CANDIDATES,
)
],
)
run_review(
cfg, console=quiet_console(), input_fn=scripted("2"), client=fixture_client()
)
(row,) = read_matches(cfg.matches_path)
assert row["match_status"] == "approved"
assert row["bgg_id"] == "205398"
assert row["year"] == "2016"
def test_reject_and_skip(tmp_path):
cfg = _setup(
tmp_path,
[
_row(title_raw="Blorvath", match_status="unmatched"),
_row(
title_raw="Citadels",
match_status="ambiguous",
candidates_json=CITADELS_CANDIDATES,
),
],
)
run_review(
cfg,
console=quiet_console(),
input_fn=scripted("r", "s"),
client=fixture_client(),
)
rows = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert rows["Blorvath"]["match_status"] == "rejected"
assert rows["Citadels"]["match_status"] == "ambiguous" # skipped, still pending
def test_unmatched_manual_id_degrades_without_token(tmp_path):
cfg = _setup(tmp_path, [_row(title_raw="Some Rare Game", match_status="unmatched")])
run_review(
cfg,
console=quiet_console(),
input_fn=scripted("m 99999"),
client=unauthorized_client(tmp_path),
)
(row,) = read_matches(cfg.matches_path)
assert row["match_status"] == "approved"
assert row["bgg_id"] == "99999"
assert row["bgg_name"] == "" # lookup blocked; id recorded anyway
def test_unmatched_research_from_fixture_cache_with_version(tmp_path):
cfg = _setup(tmp_path, [_row(title_raw="Wingspan", match_status="unmatched")])
cfg.titles_path.parent.mkdir(parents=True, exist_ok=True)
cfg.titles_path.write_text(
json.dumps(
[
{
"title_raw": "Wingspan",
"publisher_hint": "Stonemaier Games",
"year_hint": 2019,
"language_hint": "English",
"source_photos": ["x.jpg"],
}
]
)
)
run_review(
cfg,
console=quiet_console(),
input_fn=scripted("f Wingspan", "1"),
client=fixture_client(),
)
(row,) = read_matches(cfg.matches_path)
assert row["match_status"] == "approved"
assert row["bgg_id"] == "266192"
# cues + fixture versions -> resolved to the 2019 Stonemaier English edition
assert row["version_status"] == "version_auto"
assert row["version_id"] == "465063"
def test_resumable_quit_midway_then_continue(tmp_path):
rows = [
_row(
title_raw="Citadels",
match_status="ambiguous",
candidates_json=CITADELS_CANDIDATES,
),
_row(title_raw="Blorvath", match_status="unmatched"),
]
cfg = _setup(tmp_path, rows)
# first sitting: one decision, then input runs out (EOF == quit)
run_review(
cfg, console=quiet_console(), input_fn=scripted("1"), client=fixture_client()
)
by_title = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert by_title["Citadels"]["match_status"] == "approved"
assert by_title["Blorvath"]["match_status"] == "unmatched" # untouched
# second sitting resumes exactly where we left off
run_review(
cfg, console=quiet_console(), input_fn=scripted("r"), client=fixture_client()
)
by_title = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert by_title["Blorvath"]["match_status"] == "rejected"
def test_version_pass_pick_and_unknown(tmp_path):
cfg = _setup(
tmp_path,
[
_row(
title_raw="Wingspan",
match_status="auto",
bgg_id="266192",
version_status="version_ambiguous",
version_candidates_json=WINGSPAN_VERSION_CANDIDATES,
),
_row(
title_raw="Catan",
match_status="auto",
bgg_id="13",
version_status="version_ambiguous",
version_candidates_json=WINGSPAN_VERSION_CANDIDATES,
),
],
)
run_review(
cfg,
console=quiet_console(),
input_fn=scripted("y", "1", "u"),
client=fixture_client(),
)
by_title = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert by_title["Wingspan"]["version_status"] == "version_approved"
assert by_title["Wingspan"]["version_id"] == "465063"
assert by_title["Catan"]["version_status"] == "version_unknown"
assert by_title["Catan"]["version_id"] == ""
def test_version_pass_is_skippable(tmp_path):
cfg = _setup(
tmp_path,
[
_row(
title_raw="Wingspan",
match_status="auto",
bgg_id="266192",
version_status="version_ambiguous",
version_candidates_json=WINGSPAN_VERSION_CANDIDATES,
)
],
)
run_review(
cfg, console=quiet_console(), input_fn=scripted("n"), client=fixture_client()
)
(row,) = read_matches(cfg.matches_path)
assert row["version_status"] == "version_ambiguous" # untouched, review later