Eric WagonerandClaude Fable 5
1dd72d2688
Eric's question cut to the bone: "How would a user know? It matched
wiz-war and that IS the game." The auto looked unanimous because the
matcher discarded the evidence of doubt before anyone saw it — and
worse, BGG's search hides evidence of its own: results truncate
unordered in the several-hundreds (the game named "Dungeon!" appears
in NEITHER the "Dungeon!" nor the "Dungeon" search), and punctuation
can bury matches.
Three matcher changes: every title is searched raw AND depuncted,
merged by id; a name that becomes exact once its trailing
parenthetical is stripped ("Wiz-War (Eighth Edition)") is a sibling
edition — BGG files new editions as separate games — and enters the
candidate set at exact grade, so same-named lineages land in review as
a visible choice; and a LONE candidate must now earn trust (stats
fetched, sibling-grade never autos alone, true exacts must clear the
dominance ownership floor) — closing the fast path both impostors
(.dungeon at 31 owners, then Dungeon (ICP)) walked through.
Recorded outcomes: WIZ-WAR → ambiguous with all three lineages on the
ballot; Dungeon! → ambiguous (its true match is beyond BGG's search
horizon — that's what manual id is for); every legitimate auto in the
fixture set held. And the answer to Eric's second question is now
structural: re-match never re-decides — it demotes to unmatched and
the HUMAN picks from re-search or manual id; the machine only chooses
on first resolve, and it now chooses more humbly.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016jXZFSTZQKzAC8fqpWSz9g
728 lines
27 KiB
Python
728 lines
27 KiB
Python
"""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, then post-dedupes rows resolving to the same physical game
|
||
(losers become match_status="merged"; review can veto). The whole file is
|
||
rewritten atomically each run; re-runs pair entries to their existing rows
|
||
(photo overlap, then position) unless --force starts over.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import json
|
||
import re
|
||
from collections import Counter
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
import typer
|
||
from rapidfuzz import fuzz
|
||
|
||
from bggpipe.bgg_client import BGGAuthError, BGGClient, BGGQueueTimeout, client_for
|
||
from bggpipe.config import Config
|
||
from bggpipe.extract import cues_conflict, is_split, load_title_splits
|
||
from bggpipe.fsio import atomic_write_csv
|
||
from bggpipe.models import (
|
||
RECOGNIZED_MATCH_STATUSES,
|
||
GameVersion,
|
||
is_confident_version,
|
||
)
|
||
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
|
||
# "Is this the publisher on the box?" — one answer, asked in two places
|
||
# (candidate tie-break and version scoring): the sites must move together.
|
||
PUBLISHER_MATCH_THRESHOLD = 85
|
||
EDITION_NAME_THRESHOLD = 80
|
||
|
||
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",
|
||
"merged_into",
|
||
"dedupe_veto",
|
||
]
|
||
|
||
|
||
@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:
|
||
"""Cues strong enough to justify a versions fetch. Language alone
|
||
can never reach the plausibility threshold (it scores 1 of the
|
||
required 2), so fetching on it would waste a rate-limited request —
|
||
it still participates in scoring when other cues exist."""
|
||
return bool(self.publisher_hint or self.edition_hint or self.year_hint)
|
||
|
||
|
||
@dataclass
|
||
class Candidate:
|
||
bgg_id: int
|
||
name: str
|
||
year: int | None
|
||
type: str
|
||
exact: bool
|
||
fuzzy: float
|
||
sibling: bool = False # exactness earned via edition-suffix stripping
|
||
owned: int | None = None
|
||
rank: int | None = None
|
||
publishers: list[str] = field(default_factory=list)
|
||
|
||
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),
|
||
"merged_into": "",
|
||
"dedupe_veto": "",
|
||
}
|
||
|
||
|
||
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,
|
||
# always recompute: a hand-written stored value would
|
||
# silently break exact matching (both sides must normalize
|
||
# by the current rules)
|
||
title_normalized=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
|
||
|
||
|
||
_SEPARATORS = (" — ", " – ", " - ", ": ", "; ")
|
||
_GAME_WORD = re.compile(r"\s+(?:the\s+|a\s+)?game\b", re.IGNORECASE)
|
||
|
||
|
||
def _truncation_heads(title_raw: str) -> list[str]:
|
||
"""Progressively shorter heads for box titles whose printed subtitle
|
||
defeats search ("CIVILIZATION Game of the Heroic Age - ..."): text
|
||
before the first subtitle separator, before a "(The) Game ..."
|
||
descriptor, then the first two words as a last resort."""
|
||
heads: list[str] = []
|
||
present = [(title_raw.find(sep), sep) for sep in _SEPARATORS if sep in title_raw]
|
||
# earliest separator wins — priority order would let " - " late in the
|
||
# title beat an early ": ", yielding heads like "Blorvath: Quest"
|
||
sep_head = title_raw.split(min(present)[1])[0] if present else None
|
||
if sep_head:
|
||
heads.append(sep_head)
|
||
match = _GAME_WORD.search(title_raw)
|
||
if match and match.start() > 0:
|
||
heads.append(title_raw[: match.start()])
|
||
# last resort: first two words — of the pre-subtitle part, so a title
|
||
# like "Blorvath: Quest of the Zzyzx" never yields "Blorvath: Quest"
|
||
words = (sep_head or title_raw).split()
|
||
if len(words) > 2:
|
||
heads.append(" ".join(words[:2]))
|
||
|
||
seen: set[str] = {normalize_title(title_raw)}
|
||
unique: list[str] = []
|
||
for head in heads:
|
||
norm = normalize_title(head)
|
||
if norm and norm not in seen:
|
||
seen.add(norm)
|
||
unique.append(head)
|
||
return unique[:3]
|
||
|
||
|
||
def _plausible_candidates(
|
||
client: BGGClient,
|
||
entry: TitleEntry,
|
||
query: str,
|
||
head_normalized: str = "",
|
||
types: str | None = None,
|
||
) -> list[Candidate]:
|
||
"""Search BGG and keep plausible candidates, one per id: exact-normalized
|
||
or fuzzy>=90 against the FULL title, or — on truncated retries — exact
|
||
(only exact: truncation must stay conservative) against the head."""
|
||
# a fully non-Latin title normalizes to "" — empty-vs-empty is not a
|
||
# match (token_sort_ratio("", "") is 100), and searching would only
|
||
# spend rate-limited requests to prove nothing
|
||
if not entry.title_normalized:
|
||
return []
|
||
by_id: dict[int, Candidate] = {}
|
||
results = client.search(query, types) if types else client.search(query)
|
||
for result in results:
|
||
norm = normalize_title(result.name)
|
||
exact = norm == entry.title_normalized
|
||
# BGG files new editions as SEPARATE games named "X (Nth Edition)":
|
||
# a name that equals the title once its trailing parenthetical is
|
||
# stripped is a sibling edition — exact-grade, or the match looks
|
||
# unanimously confident while hiding the real choice (Wiz-War has
|
||
# three same-named lineages; the spec's top failure mode)
|
||
sibling = False
|
||
if not exact and result.type != "boardgameexpansion":
|
||
sibling = (
|
||
normalize_title(_EDITION_SUFFIX.sub("", result.name))
|
||
== entry.title_normalized
|
||
)
|
||
exact = sibling
|
||
fuzzy = fuzz.token_sort_ratio(norm, entry.title_normalized)
|
||
if not exact and fuzzy < FUZZY_THRESHOLD:
|
||
if not (head_normalized and norm == head_normalized):
|
||
continue
|
||
exact = True # head-exact counts as strong, nothing weaker does
|
||
candidate = Candidate(
|
||
bgg_id=result.bgg_id,
|
||
name=result.name,
|
||
year=result.year,
|
||
type=result.type,
|
||
exact=exact,
|
||
sibling=sibling,
|
||
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:
|
||
# a lone candidate must still earn trust: BGG's search visibly
|
||
# truncates generic queries (the game named "Dungeon!" appears in
|
||
# NEITHER of its own searches), so the sole survivor may be an
|
||
# impostor standing where the famous game should be. Sibling-grade
|
||
# exactness never autos alone, and a true exact must clear the
|
||
# same ownership floor the dominance rule enforces.
|
||
only = cands[0]
|
||
stats = {t.bgg_id: t for t in client.things([only.bgg_id], stats=True)}
|
||
if only.bgg_id in stats:
|
||
only.owned = stats[only.bgg_id].owned
|
||
only.rank = stats[only.bgg_id].rank
|
||
only.publishers = list(stats[only.bgg_id].publishers)
|
||
row.candidates = [only]
|
||
if only.sibling or not only.exact or (only.owned or 0) < DOMINANCE_MIN_OWNED:
|
||
row.match_status = "ambiguous"
|
||
return row
|
||
chosen = only
|
||
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
|
||
c.publishers = list(stats[c.bgg_id].publishers)
|
||
row.candidates = top
|
||
chosen = _publisher_pick(entry, top) or _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 _publisher_pick(entry: TitleEntry, top: list[Candidate]) -> Candidate | None:
|
||
"""When a publisher was legible on the box, and exactly one exact-named
|
||
candidate is from that publisher, that's the game."""
|
||
if not entry.publisher_hint:
|
||
return None
|
||
if len({c.type for c in top}) > 1:
|
||
# mixed base-game/expansion candidates are never auto-resolved —
|
||
# the same veto _dominant applies (spec's top failure mode); an
|
||
# alternate name can make an expansion "exact" too
|
||
return None
|
||
hint = normalize_title(entry.publisher_hint)
|
||
matches = [
|
||
c
|
||
for c in top
|
||
if c.exact
|
||
and any(
|
||
fuzz.partial_ratio(hint, normalize_title(p)) >= PUBLISHER_MATCH_THRESHOLD
|
||
for p in c.publishers
|
||
)
|
||
]
|
||
return matches[0] if len(matches) == 1 else None
|
||
|
||
|
||
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)) >= PUBLISHER_MATCH_THRESHOLD
|
||
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)
|
||
)
|
||
>= EDITION_NAME_THRESHOLD
|
||
):
|
||
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).
|
||
Public: review's manual-id path calls this too."""
|
||
if not entry.has_version_cues:
|
||
row.version_status = "version_unknown"
|
||
return
|
||
things = client.things([row.bgg_id], versions=True)
|
||
if not things:
|
||
# a manually-typed id BGG doesn't know: nothing to offer
|
||
row.version_status = "version_unknown"
|
||
return
|
||
thing = things[0]
|
||
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 # every plausible version: no silent cap
|
||
]
|
||
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"
|
||
|
||
|
||
_EDITION_SUFFIX = re.compile(r"\s*\([^)]*\)\s*$")
|
||
_PUNCT = re.compile(r"[^\w\s]", re.UNICODE)
|
||
|
||
|
||
def _depunct(title: str) -> str:
|
||
"""BGG's search engine can choke on punctuation — the game literally
|
||
named "Dungeon!" is missing from its own 824-result search. A plain
|
||
query recovers it."""
|
||
return " ".join(_PUNCT.sub(" ", title).split())
|
||
|
||
|
||
def _merged_candidates(
|
||
client: BGGClient, entry: TitleEntry, query: str, types: str | None = None
|
||
) -> list[Candidate]:
|
||
"""Search the raw title AND its punctuation-free form, merged by id
|
||
(strongest evidence wins)."""
|
||
cands = _plausible_candidates(client, entry, query, types=types)
|
||
plain = _depunct(query)
|
||
if plain.casefold() != query.casefold():
|
||
by_id = {c.bgg_id: c for c in cands}
|
||
for c in _plausible_candidates(client, entry, plain, types=types):
|
||
prev = by_id.get(c.bgg_id)
|
||
if prev is None or (c.exact, c.fuzzy) > (prev.exact, prev.fuzzy):
|
||
by_id[c.bgg_id] = c
|
||
cands = sorted(by_id.values(), key=lambda c: (not c.exact, -c.fuzzy))
|
||
return cands
|
||
|
||
|
||
def resolve_entry(client: BGGClient, entry: TitleEntry) -> MatchRow:
|
||
cands = _merged_candidates(client, entry, entry.title_raw)
|
||
if not cands:
|
||
# Long transcribed box titles ("CIVILIZATION Game of the Heroic
|
||
# Age - ...") defeat search: retry with progressively shorter heads
|
||
# instead of loosening the fuzzy threshold.
|
||
for head in _truncation_heads(entry.title_raw):
|
||
cands = _plausible_candidates(client, entry, head, normalize_title(head))
|
||
if cands:
|
||
break
|
||
if not cands:
|
||
# Not a board game? RPGs live in the same geekdo database under
|
||
# type=rpgitem (same API, same token). A hit becomes a LOCAL
|
||
# library citizen: identified and enriched, never uploaded (diff
|
||
# routes rpgitem rows to local_only).
|
||
cands = _merged_candidates(client, entry, entry.title_raw, types="rpgitem")
|
||
if not cands:
|
||
for head in _truncation_heads(entry.title_raw):
|
||
cands = _plausible_candidates(
|
||
client, entry, head, normalize_title(head), types="rpgitem"
|
||
)
|
||
if cands:
|
||
break
|
||
row = _classify(client, entry, cands)
|
||
if row.match_status == "auto":
|
||
resolve_version(client, entry, row)
|
||
return row
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MergeEvent:
|
||
loser_title: str
|
||
survivor_title: str
|
||
bgg_name: str
|
||
bgg_id: str
|
||
|
||
|
||
def dedupe_matches(
|
||
rows: list[dict],
|
||
titles: list[TitleEntry],
|
||
splits: list[dict] | None = None,
|
||
) -> list[MergeEvent]:
|
||
"""Post-resolve dedupe: rows resolving to the same (bgg_id, version_id —
|
||
or both version-unknown) are the same physical game seen twice (a typo
|
||
read, a partial spine) UNLESS their extraction cues conflict, which
|
||
means two editions. Losers are marked match_status="merged" pointing at
|
||
the survivor via merged_into — no row is ever deleted, and the review
|
||
UI can veto the merge."""
|
||
entry_by_key = {(e.title_raw, ";".join(e.source_photos)): e for e in titles}
|
||
entry_by_title: dict[str, TitleEntry] = {}
|
||
for e in titles:
|
||
entry_by_title.setdefault(e.title_raw, e)
|
||
|
||
def cues(row: dict) -> dict:
|
||
entry = entry_by_key.get(
|
||
(row["title_raw"], row["source_photos"])
|
||
) or entry_by_title.get(row["title_raw"])
|
||
if entry is None:
|
||
return {}
|
||
return {
|
||
"publisher_hint": entry.publisher_hint,
|
||
"edition_hint": entry.edition_hint,
|
||
"language_hint": entry.language_hint,
|
||
"year_hint": entry.year_hint,
|
||
}
|
||
|
||
groups: dict[tuple[str, str], list[dict]] = {}
|
||
for row in rows:
|
||
if row["match_status"] not in RECOGNIZED_MATCH_STATUSES or not row["bgg_id"]:
|
||
continue
|
||
if row.get("dedupe_veto"):
|
||
# a human already ruled "this is a genuinely separate copy" —
|
||
# re-running resolve must never overturn that (spec: re-runs
|
||
# lose no work, least of all review decisions)
|
||
continue
|
||
if is_split(
|
||
normalize_title(row["title_raw"]),
|
||
row["source_photos"].split(";"),
|
||
splits or [],
|
||
):
|
||
continue # human-split copies: per-photo rows stay separate
|
||
key = (
|
||
row["bgg_id"],
|
||
row["version_id"] if is_confident_version(row) else "",
|
||
)
|
||
groups.setdefault(key, []).append(row)
|
||
|
||
events: list[MergeEvent] = []
|
||
for (bgg_id, _version), group in groups.items():
|
||
if len(group) < 2:
|
||
continue
|
||
# survivor: the row whose transcription best matches the BGG name
|
||
group = sorted(
|
||
group,
|
||
key=lambda r: (
|
||
normalize_title(r["title_raw"]) != normalize_title(r["bgg_name"] or "")
|
||
),
|
||
)
|
||
survivor = group[0]
|
||
for loser in group[1:]:
|
||
if cues_conflict(cues(survivor), cues(loser)):
|
||
continue # conflicting edition cues: genuinely two copies
|
||
loser["match_status"] = "merged"
|
||
loser["merged_into"] = survivor["title_raw"]
|
||
events.append(
|
||
MergeEvent(
|
||
loser_title=loser["title_raw"],
|
||
survivor_title=survivor["title_raw"],
|
||
bgg_name=survivor["bgg_name"],
|
||
bgg_id=bgg_id,
|
||
)
|
||
)
|
||
# A survivor can itself lose in a later run; rows pointing at it would
|
||
# form a chain diff's one-level photo hop can't follow. Rewrite every
|
||
# merged row to its terminal survivor.
|
||
status_by_title = {r["title_raw"]: r for r in rows}
|
||
for row in rows:
|
||
if row["match_status"] != "merged":
|
||
continue
|
||
target, hops = row["merged_into"], 0
|
||
while (
|
||
hops < 10
|
||
and (nxt := status_by_title.get(target)) is not None
|
||
and nxt["match_status"] == "merged"
|
||
and nxt["merged_into"]
|
||
):
|
||
target = nxt["merged_into"]
|
||
hops += 1
|
||
row["merged_into"] = target
|
||
return events
|
||
|
||
|
||
def read_matches(path: Path) -> list[dict[str, str]]:
|
||
if not path.exists():
|
||
return []
|
||
with path.open(newline="") as f:
|
||
rows = list(csv.DictReader(f))
|
||
for row in rows: # optional columns: tolerate rows without them
|
||
row.setdefault("merged_into", "")
|
||
row.setdefault("dedupe_veto", "")
|
||
return rows
|
||
|
||
|
||
def write_matches(path: Path, rows: list[dict[str, str]]) -> int:
|
||
"""Atomic full rewrite — review updates rows in place decision by
|
||
decision. Returns the written file's mtime_ns so the caller can record
|
||
its own write without a re-stat race."""
|
||
return atomic_write_csv(path, MATCH_COLUMNS, rows)
|
||
|
||
|
||
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_rows = read_matches(cfg.matches_path)
|
||
client = client or client_for(cfg)
|
||
|
||
# Pair entries with existing rows BY TITLE: photo-overlap first, then
|
||
# position. Pure position breaks when titles.json order churns (a
|
||
# reshoot photo sorting earlier reorders same-title entries); overlap
|
||
# keeps each edition glued to its own row, and position only settles
|
||
# entries with no photo history.
|
||
rows_by_title: dict[str, list[dict]] = {}
|
||
for row in existing_rows:
|
||
rows_by_title.setdefault(row["title_raw"], []).append(row)
|
||
|
||
claimed_rows: set[int] = set()
|
||
|
||
def pair_row(entry: TitleEntry) -> dict | None:
|
||
candidates = [
|
||
r
|
||
for r in rows_by_title.get(entry.title_raw, [])
|
||
if id(r) not in claimed_rows
|
||
]
|
||
if not candidates:
|
||
return None
|
||
photos = set(entry.source_photos)
|
||
for r in candidates:
|
||
if photos & set(r["source_photos"].split(";")):
|
||
claimed_rows.add(id(r))
|
||
return r
|
||
return None
|
||
|
||
def pair_row_positional(entry: TitleEntry) -> dict | None:
|
||
candidates = [
|
||
r
|
||
for r in rows_by_title.get(entry.title_raw, [])
|
||
if id(r) not in claimed_rows
|
||
]
|
||
if candidates:
|
||
claimed_rows.add(id(candidates[0]))
|
||
return candidates[0]
|
||
return None
|
||
|
||
new_rows: list[MatchRow] = []
|
||
skipped = 0
|
||
photos_updated = False
|
||
blocked: list[str] = []
|
||
blocked_titles: set[str] = set()
|
||
# overlap pass first so a reordered titles.json can't mispair editions
|
||
paired_by_id: dict[int, dict] = {}
|
||
for entry in entries:
|
||
row_dict = pair_row(entry)
|
||
if row_dict is not None:
|
||
paired_by_id[id(entry)] = row_dict
|
||
for entry in entries:
|
||
if id(entry) not in paired_by_id:
|
||
row_dict = pair_row_positional(entry)
|
||
if row_dict is not None:
|
||
paired_by_id[id(entry)] = row_dict
|
||
|
||
for entry in entries:
|
||
row_dict = paired_by_id.get(id(entry))
|
||
if row_dict is not None:
|
||
photos = ";".join(entry.source_photos)
|
||
if row_dict["source_photos"] != photos and not row_dict.get("dedupe_veto"):
|
||
# provenance follows the entry — except on split/vetoed rows,
|
||
# whose per-copy photo sets are human-authored
|
||
row_dict["source_photos"] = photos
|
||
photos_updated = True
|
||
skipped += 1
|
||
continue
|
||
if entry.title_raw in blocked_titles:
|
||
# an earlier same-title entry is waiting on the token: resolving
|
||
# this one now would claim the wrong pairing slot on the next
|
||
# run — defer the whole group
|
||
blocked.append(entry.title_raw)
|
||
continue
|
||
try:
|
||
row = resolve_entry(client, entry)
|
||
except (BGGAuthError, BGGQueueTimeout) as err:
|
||
# No token / BGG still queueing: cached titles still resolve;
|
||
# the rest wait. No row is written, so a future run picks them
|
||
# up untouched.
|
||
blocked.append(entry.title_raw)
|
||
blocked_titles.add(entry.title_raw)
|
||
reason = (
|
||
"waiting on BGG API token"
|
||
if isinstance(err, BGGAuthError)
|
||
else "BGG still queueing — re-run in a minute"
|
||
)
|
||
typer.echo(f" {entry.title_raw!r} -> {reason}")
|
||
continue
|
||
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}")
|
||
|
||
all_rows = existing_rows + [row.to_csv() for row in new_rows]
|
||
merges = dedupe_matches(all_rows, entries, load_title_splits(cfg.title_splits_path))
|
||
if new_rows or photos_updated or merges:
|
||
write_matches(cfg.matches_path, all_rows) # atomic full rewrite
|
||
if merges:
|
||
typer.echo("")
|
||
for m in merges:
|
||
typer.echo(
|
||
f" merged {m.loser_title!r} into {m.survivor_title!r} — same "
|
||
f"game ({m.bgg_name}, {m.bgg_id}); veto in review if wrong"
|
||
)
|
||
|
||
counts = Counter(row.match_status for row in new_rows)
|
||
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}."
|
||
)
|
||
if blocked:
|
||
typer.echo(
|
||
f"\n{len(blocked)} title(s) are waiting on the BGG API "
|
||
"(register at https://boardgamegeek.com/applications, then set "
|
||
"BGG_API_TOKEN and re-run resolve — everything above is saved): "
|
||
+ ", ".join(blocked)
|
||
)
|
||
return new_rows
|