Diff stage: snapshot/live collection modes, to_add + to_update outputs

compute_diff is a pure function over matches + collection items: new
games land in to_add.csv (carrying a confident version when matching
produced one); owned version-less entries with version_auto/approved
matches produce additive to_update.csv rows keyed by collid, consuming
distinct collids when photos show two editions; entries that already
carry a version are never touched — disagreements are reported in the
summary. Unseen collection entries are listed informationally. Live API
mode activates when BGG_API_TOKEN + username exist; otherwise the two
hand-pulled snapshot XMLs (real 2018 collection, 79 unique copies after
collid dedupe) are used, and they double as parsing fixtures in tests.

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

src/bggpipe/cli.py

@@ -67,7 +67,10 @@ def review(config: ConfigOpt = None) -> None:
 @app.command()
 def diff(config: ConfigOpt = None) -> None:
     """Stage 4: diff approved matches against the existing BGG collection."""
-    _not_implemented("diff", 4)
+    from bggpipe.diff import run_diff
+
+    cfg = load_config(config)
+    run_diff(cfg)
 
 
 @app.command()

src/bggpipe/diff.py

@@ -0,0 +1,213 @@
+"""Stage 4 — diff approved/auto matches against the existing BGG collection.
+
+Two collection sources:
+- live API (default once BGG_API_TOKEN exists): /collection base + expansion
+  subtype calls, merged by collid;
+- snapshot files (fallback): data/collection_snapshot_base.xml +
+  data/collection_snapshot_expansions.xml, pulled manually via the
+  logged-in-user exemption.
+
+Outputs both artifacts:
+- to_add.csv    — recognized games not in the collection (new entries);
+- to_update.csv — owned, VERSION-LESS entries where matching produced a
+  confident version (version_auto/version_approved). Strictly additive:
+  entries that already carry a version are never touched — a photo/version
+  disagreement is reported in the summary instead.
+"""
+
+from __future__ import annotations
+
+import csv
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import typer
+
+from bggpipe.bgg_client import BGGClient
+from bggpipe.config import Config
+from bggpipe.models import CollectionItem, parse_collection
+from bggpipe.resolve import read_matches
+
+TO_ADD_COLUMNS = [
+    "bgg_id",
+    "bgg_name",
+    "year",
+    "type",
+    "version_id",
+    "version_name",
+    "title_raw",
+    "source_photos",
+]
+TO_UPDATE_COLUMNS = ["collid", "bgg_id", "bgg_name", "version_id", "version_name"]
+
+SNAPSHOT_FILES = ("collection_snapshot_base.xml", "collection_snapshot_expansions.xml")
+_CONFIDENT_VERSION = ("version_auto", "version_approved")
+
+
+@dataclass
+class DiffResult:
+    to_add: list[dict] = field(default_factory=list)
+    to_update: list[dict] = field(default_factory=list)
+    already_owned: list[str] = field(default_factory=list)  # title_raw
+    disagreements: list[str] = field(default_factory=list)
+    unseen: list[CollectionItem] = field(default_factory=list)
+    pending: list[str] = field(default_factory=list)  # ambiguous/unmatched titles
+    rejected: int = 0
+    recognized: int = 0
+
+
+def load_snapshot_collection(data_dir: Path) -> list[CollectionItem]:
+    """Merge the base + expansions snapshot files, deduping by collid (the
+    same physical copy can appear in both responses)."""
+    items: list[CollectionItem] = []
+    seen: set[int] = set()
+    for name in SNAPSHOT_FILES:
+        path = data_dir / name
+        if not path.exists():
+            raise FileNotFoundError(
+                f"{path} not found — pull your collection while logged in "
+                "(no registration needed for your own collection) or set "
+                "BGG_API_TOKEN for live mode."
+            )
+        for item in parse_collection(path.read_text()):
+            if item.own and item.coll_id not in seen:
+                seen.add(item.coll_id)
+                items.append(item)
+    return items
+
+
+def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResult:
+    by_object: dict[int, list[CollectionItem]] = {}
+    for item in collection:
+        by_object.setdefault(item.object_id, []).append(item)
+
+    result = DiffResult()
+    seen_object_ids: set[int] = set()
+    consumed_collids: set[int] = set()
+
+    for row in rows:
+        status = row["match_status"]
+        if status == "rejected":
+            result.rejected += 1
+            continue
+        if status not in ("auto", "approved") or not row["bgg_id"]:
+            result.pending.append(row["title_raw"])
+            continue
+
+        result.recognized += 1
+        bgg_id = int(row["bgg_id"])
+        copies = by_object.get(bgg_id, [])
+        if copies:
+            seen_object_ids.add(bgg_id)
+
+        confident = row["version_status"] in _CONFIDENT_VERSION and row["version_id"]
+        version_id = int(row["version_id"]) if confident else None
+
+        if not copies:
+            result.to_add.append(
+                {
+                    "bgg_id": row["bgg_id"],
+                    "bgg_name": row["bgg_name"],
+                    "year": row["year"],
+                    "type": row["type"],
+                    "version_id": row["version_id"] if confident else "",
+                    "version_name": row["version_name"] if confident else "",
+                    "title_raw": row["title_raw"],
+                    "source_photos": row["source_photos"],
+                }
+            )
+            continue
+
+        if not confident:
+            # bare id with unknown version: owned if any copy exists
+            result.already_owned.append(row["title_raw"])
+            continue
+
+        if any(c.version_id == version_id for c in copies):
+            result.already_owned.append(row["title_raw"])
+            continue
+
+        versionless = [
+            c
+            for c in copies
+            if c.version_id is None and c.coll_id not in consumed_collids
+        ]
+        if versionless:
+            target = versionless[0]
+            consumed_collids.add(target.coll_id)
+            result.already_owned.append(row["title_raw"])
+            result.to_update.append(
+                {
+                    "collid": str(target.coll_id),
+                    "bgg_id": row["bgg_id"],
+                    "bgg_name": row["bgg_name"] or target.name,
+                    "version_id": row["version_id"],
+                    "version_name": row["version_name"],
+                }
+            )
+        else:
+            # every copy already carries a (different) version — never touch it
+            result.already_owned.append(row["title_raw"])
+            result.disagreements.append(
+                f"{row['title_raw']}: photo suggests version "
+                f"{row['version_name']!r} ({row['version_id']}) but the "
+                "collection entry already has a version set — not changing it"
+            )
+
+    result.unseen = [
+        item for item in collection if item.object_id not in seen_object_ids
+    ]
+    return result
+
+
+def _write_csv(path: Path, columns: list[str], rows: list[dict]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=columns, extrasaction="ignore")
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def run_diff(cfg: Config, *, client: BGGClient | None = None) -> DiffResult:
+    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)
+
+    if os.environ.get("BGG_API_TOKEN") and cfg.bgg_username:
+        typer.echo("Fetching live collection from BGG…")
+        client = client or BGGClient(cfg.cache_dir, cfg.rate_limit_seconds)
+        collection = client.collection_full(cfg.bgg_username)
+    else:
+        typer.echo(
+            "No BGG_API_TOKEN — using collection snapshot files in "
+            f"{cfg.data_dir}/ (live mode takes over once the token exists)."
+        )
+        collection = load_snapshot_collection(cfg.data_dir)
+
+    result = compute_diff(rows, collection)
+
+    _write_csv(cfg.data_dir / "to_add.csv", TO_ADD_COLUMNS, result.to_add)
+    _write_csv(cfg.data_dir / "to_update.csv", TO_UPDATE_COLUMNS, result.to_update)
+
+    typer.echo(
+        f"\n{result.recognized} recognized · {len(result.already_owned)} already "
+        f"owned · {len(result.to_add)} to add · {len(result.to_update)} version "
+        f"update(s) · {len(result.pending)} pending review · "
+        f"{result.rejected} rejected"
+    )
+    if result.disagreements:
+        typer.echo("\nVersion disagreements (left untouched):")
+        for line in result.disagreements:
+            typer.echo(f"  - {line}")
+    if result.pending:
+        typer.echo("\nStill pending review: " + ", ".join(result.pending))
+    if result.unseen:
+        typer.echo(
+            f"\nIn your collection but not seen in any photo "
+            f"({len(result.unseen)} — informational only):"
+        )
+        for item in result.unseen:
+            typer.echo(f"  - {item.name} ({item.object_id})")
+    return result