Stage 5 upload: Playwright-driven adds and version updates

Queue from to_add/to_update minus upload_log.csv (append-per-attempt,
so runs resume); per-game failure isolation with 2-4s pacing;
--dry-run/--verify/--retry-failed/--limit; stub-fixture marker blocks
real runs, dry-run warns. Headed browser by default: live recon showed
Cloudflare Turnstile hard-blocks headless, and BGG never reaches
networkidle. Login selectors verified anonymously; version-picker
pagination and the collection-row update flow remain unverified until
real data exists. Client collection fetches gain a refresh passthrough
so --verify sees the live collection, not cache.

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

src/bggpipe/bgg_client.py

@@ -157,17 +157,25 @@ class BGGClient:
         username: str,
         subtype: str | None = None,
         version: bool = True,
+        *,
+        refresh: bool = False,
     ) -> list[CollectionItem]:
         params = {"username": username, "own": "1"}
         if subtype:
             params["subtype"] = subtype
         if version:
             params["version"] = "1"
-        return parse_collection(self.get_xml("collection", params))
+        return parse_collection(self.get_xml("collection", params, refresh=refresh))
 
-    def collection_full(self, username: str) -> list[CollectionItem]:
-        """Owned items incl. expansions (excluded from the default subtype)."""
-        base = self.collection(username)
-        expansions = self.collection(username, subtype="boardgameexpansion")
+    def collection_full(
+        self, username: str, *, refresh: bool = False
+    ) -> list[CollectionItem]:
+        """Owned items incl. expansions (excluded from the default subtype).
+        refresh=True bypasses the cache — upload --verify must see the live
+        collection, not the snapshot resolve ran against."""
+        base = self.collection(username, refresh=refresh)
+        expansions = self.collection(
+            username, subtype="boardgameexpansion", refresh=refresh
+        )
         seen = {item.coll_id for item in base}
         return base + [e for e in expansions if e.coll_id not in seen]

src/bggpipe/cli.py

@@ -20,13 +20,6 @@ ConfigOpt = Annotated[
 ]
 
 
-def _not_implemented(stage: str, build_order: int) -> None:
-    typer.echo(
-        f"bggpipe {stage}: not implemented yet (build-order step {build_order})."
-    )
-    raise typer.Exit(code=1)
-
-
 @app.command()
 def extract(
     only: Annotated[
@@ -89,13 +82,39 @@ def diff(config: ConfigOpt = None) -> None:
 
 @app.command()
 def upload(
-    dry_run: Annotated[bool, typer.Option("--dry-run")] = False,
-    verify: Annotated[bool, typer.Option("--verify")] = False,
-    retry_failed: Annotated[bool, typer.Option("--retry-failed")] = False,
+    dry_run: Annotated[
+        bool, typer.Option("--dry-run", help="Show the queue without a browser")
+    ] = False,
+    verify: Annotated[
+        bool, typer.Option("--verify", help="Re-fetch the collection and cross-check")
+    ] = False,
+    retry_failed: Annotated[
+        bool, typer.Option("--retry-failed", help="Re-attempt previously failed games")
+    ] = False,
+    limit: Annotated[
+        int | None, typer.Option("--limit", help="Upload at most N games this run")
+    ] = None,
+    headless: Annotated[
+        bool,
+        typer.Option(
+            "--headless",
+            help="Run the browser headless (Cloudflare may block login)",
+        ),
+    ] = False,
     config: ConfigOpt = None,
 ) -> None:
     """Stage 5: add games to the BGG collection via Playwright."""
-    _not_implemented("upload", 5)
+    from bggpipe.upload import run_upload
+
+    cfg = load_config(config)
+    run_upload(
+        cfg,
+        dry_run=dry_run,
+        verify=verify,
+        retry_failed=retry_failed,
+        limit=limit,
+        headless=headless,
+    )
 
 
 @app.command()

src/bggpipe/upload.py

@@ -0,0 +1,507 @@
+"""Stage 5 — upload: add games / set versions on boardgamegeek.com via Playwright.
+
+BGG has no write API, so this stage drives the real website with a logged-in
+browser session. Etiquette (spec + bgg-api skill):
+- 2-4 s randomized delay between games;
+- credentials only from BGG_USERNAME / BGG_PASSWORD env vars, never disk/logs;
+- browser storage state persists locally (gitignored) so login is rare;
+- every attempt is appended to data/upload_log.csv immediately, so a killed
+  run loses nothing and re-runs skip completed work;
+- refuses to touch the site while data/bgg_cache/STUB_FIXTURES.marker exists
+  (stub-resolved version ids must never reach BGG); --dry-run still works,
+  loudly labeled as synthetic.
+
+Cloudflare: BGG fronts the site with a Turnstile check that blocks headless
+browsers outright (verified 2026-08-01 — headless shell never gets past
+"Just a moment..."). The stage therefore runs HEADED by default; a first
+login may need one human click on the challenge widget. BGG pages also never
+reach Playwright's networkidle (ad/analytics polling), so navigation waits on
+domcontentloaded plus explicit element waits.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import csv
+import os
+import random
+import re
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Protocol
+
+import typer
+
+from bggpipe.bgg_client import BGGAuthError, BGGClient
+from bggpipe.config import Config
+from bggpipe.models import CollectionItem
+
+BGG = "https://boardgamegeek.com"
+STORAGE_STATE_PATH = Path("storage_state.json")  # gitignored, credential-adjacent
+UPLOAD_LOG_COLUMNS = [
+    "action",
+    "bgg_id",
+    "collid",
+    "name",
+    "version_id",
+    "status",
+    "timestamp",
+    "error",
+]
+DONE_STATUSES = {"added", "updated", "already_present"}
+MAX_VERSION_PAGES = 40
+
+
+@dataclass(frozen=True)
+class UploadJob:
+    action: str  # "add" | "update"
+    bgg_id: str
+    name: str
+    version_id: str = ""
+    version_name: str = ""
+    collid: str = ""
+
+    @property
+    def key(self) -> tuple[str, str, str]:
+        # A second copy of the same game (different version) is a distinct
+        # add; updates are keyed by the physical copy they amend.
+        if self.action == "update":
+            return ("update", self.collid, "")
+        return ("add", self.bgg_id, self.version_id)
+
+
+def _row_key(row: dict) -> tuple[str, str, str]:
+    if row["action"] == "update":
+        return ("update", row["collid"], "")
+    return ("add", row["bgg_id"], row["version_id"])
+
+
+def _read_csv(path: Path) -> list[dict]:
+    if not path.exists():
+        return []
+    with path.open(newline="") as f:
+        return list(csv.DictReader(f))
+
+
+def append_log_row(path: Path, row: dict) -> None:
+    """Append one attempt, creating the file with a header on first write.
+    One row per attempt, flushed immediately — the log is the resume point."""
+    new = not path.exists()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=UPLOAD_LOG_COLUMNS, extrasaction="ignore")
+        if new:
+            writer.writeheader()
+        writer.writerow(row)
+
+
+def build_queue(
+    to_add: list[dict],
+    to_update: list[dict],
+    log_rows: list[dict],
+    *,
+    retry_failed: bool = False,
+) -> tuple[list[UploadJob], int, int]:
+    """Turn the diff outputs into pending jobs, minus work the log says is
+    done. Returns (jobs, skipped_done, skipped_failed)."""
+    latest: dict[tuple[str, str, str], str] = {}
+    for row in log_rows:
+        latest[_row_key(row)] = row["status"]
+
+    candidates = [
+        UploadJob(
+            action="add",
+            bgg_id=row["bgg_id"],
+            name=row["bgg_name"],
+            version_id=row.get("version_id", ""),
+            version_name=row.get("version_name", ""),
+        )
+        for row in to_add
+    ] + [
+        UploadJob(
+            action="update",
+            bgg_id=row["bgg_id"],
+            name=row["bgg_name"],
+            version_id=row["version_id"],
+            version_name=row["version_name"],
+            collid=row["collid"],
+        )
+        for row in to_update
+    ]
+
+    jobs: list[UploadJob] = []
+    skipped_done = skipped_failed = 0
+    for job in candidates:
+        status = latest.get(job.key)
+        if status in DONE_STATUSES:
+            skipped_done += 1
+        elif status == "failed" and not retry_failed:
+            skipped_failed += 1
+        else:
+            jobs.append(job)
+    return jobs, skipped_done, skipped_failed
+
+
+def _scrub(text: str) -> str:
+    """Credentials must never reach the log, even via a selector error that
+    echoes filled form values."""
+    for key in ("BGG_PASSWORD", "BGG_USERNAME"):
+        value = os.environ.get(key)
+        if value:
+            text = text.replace(value, "***")
+    return text
+
+
+class Uploader(Protocol):
+    def add_game(self, job: UploadJob) -> tuple[str, str]: ...
+
+    def update_entry(self, job: UploadJob) -> tuple[str, str]: ...
+
+
+class PlaywrightUploader:
+    """Drives the real site. Selector documentation: docs/bgg-upload-flow.md.
+
+    Verified selectors (2026-08-01): the login form (#inputUsername /
+    #inputPassword / "Sign In") and the Add-to-Collection dialog structure.
+    UNVERIFIED: version-picker pagination, post-save behavior, and the whole
+    collection-row edit flow for updates — expect first-real-run adjustments.
+    """
+
+    def __init__(
+        self,
+        username: str,
+        storage_state: Path = STORAGE_STATE_PATH,
+        headless: bool = False,
+    ) -> None:
+        self._username = username
+        self._storage_state = storage_state
+        self._headless = headless
+        self._authed = False
+
+    def __enter__(self) -> PlaywrightUploader:
+        from playwright.sync_api import TimeoutError as PWTimeoutError
+        from playwright.sync_api import sync_playwright
+
+        self._timeout_error = PWTimeoutError
+        self._pw = sync_playwright().start()
+        self._browser = self._pw.chromium.launch(headless=self._headless)
+        state = str(self._storage_state) if self._storage_state.exists() else None
+        self._context = self._browser.new_context(storage_state=state)
+        self._page = self._context.new_page()
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        self._context.close()
+        self._browser.close()
+        self._pw.stop()
+
+    def _goto(self, url: str) -> None:
+        # networkidle never arrives on BGG (ad polling) — domcontentloaded
+        # plus per-element auto-waiting is the reliable pattern.
+        self._page.goto(url, wait_until="domcontentloaded")
+
+    def _signed_out(self) -> bool:
+        # Heuristic: the header shows a "Sign In" link only when logged out.
+        return (
+            self._page.get_by_role("link", name=re.compile(r"^sign in$", re.I)).count()
+            > 0
+        )
+
+    def _ensure_logged_in(self) -> None:
+        if self._authed:
+            return
+        page = self._page
+        self._goto(f"{BGG}/")
+        if self._signed_out():
+            user = os.environ.get("BGG_USERNAME", "")
+            password = os.environ.get("BGG_PASSWORD", "")
+            if not (user and password):
+                raise RuntimeError(
+                    "BGG_USERNAME and BGG_PASSWORD env vars are required to log in"
+                )
+            self._goto(f"{BGG}/login")
+            # Generous timeout: in headed mode a human may need to click the
+            # Cloudflare Turnstile widget before the form renders.
+            page.locator("#inputUsername").wait_for(timeout=120_000)
+            page.locator("#inputUsername").fill(user)
+            page.locator("#inputPassword").fill(password)
+            page.get_by_role("button", name="Sign In").click()
+            page.wait_for_url(lambda url: "/login" not in url, timeout=120_000)
+            self._context.storage_state(path=str(self._storage_state))
+        self._authed = True
+
+    def _open_dialog(self, opener) -> object:
+        """Click an opener that may no-op right after page load (hydration
+        race) and wait for the dialog to actually show."""
+        dialog = self._page.get_by_role("dialog")
+        for attempt in (1, 2):
+            opener.click()
+            try:
+                dialog.wait_for(state="visible", timeout=5_000)
+                return dialog
+            except self._timeout_error:
+                if attempt == 2:
+                    raise
+        return dialog
+
+    def _select_version(self, dialog, version_name: str) -> bool:
+        """Page through the version sub-view matching the canonical version
+        NAME (the list has no search box). Returns False — with the sub-view
+        cancelled — when the name never shows up."""
+        dialog.get_by_role("button", name="Set version/edition").click()
+        pattern = re.compile(re.escape(version_name), re.I)
+        with contextlib.suppress(self._timeout_error):  # empty list is legal
+            dialog.get_by_role("listitem").first.wait_for(timeout=10_000)
+        for _ in range(MAX_VERSION_PAGES):
+            items = dialog.get_by_role("listitem").filter(has_text=pattern)
+            if items.count():
+                items.first.click()
+                return True
+            # Pagination controls are UNVERIFIED (docs/bgg-upload-flow.md);
+            # best guess is a next-page button, stopping when absent/disabled.
+            nxt = dialog.get_by_role("button", name=re.compile("next|›|»", re.I)).first
+            if nxt.count() == 0 or nxt.is_disabled():
+                break
+            nxt.click()
+        # Two-level dismissal: the sub-view has its own Cancel distinct from
+        # the main dialog's.
+        dialog.get_by_role("button", name="Cancel").first.click()
+        return False
+
+    def add_game(self, job: UploadJob) -> tuple[str, str]:
+        self._ensure_logged_in()
+        page = self._page
+        # /boardgame/<id> redirects to the canonical slug for any subtype.
+        self._goto(f"{BGG}/boardgame/{job.bgg_id}/")
+        add_btn = page.get_by_role("button", name="Add To").first
+        dialog = self._open_dialog(add_btn)
+        # Content settles when the game-name heading replaces "Loading...".
+        dialog.get_by_role(
+            "heading", name=re.compile(re.escape(job.name), re.I)
+        ).wait_for(timeout=15_000)
+        dialog.get_by_label("Own").check()
+        note = ""
+        if job.version_name and not self._select_version(dialog, job.version_name):
+            note = f"version {job.version_name!r} not in picker; added without version"
+        dialog.get_by_role("button", name="Save").click()
+        # The dialog is hidden after save, not removed from the DOM.
+        dialog.wait_for(state="hidden", timeout=15_000)
+        return "added", note
+
+    def update_entry(self, job: UploadJob) -> tuple[str, str]:
+        """Set the version on an EXISTING entry — strictly additive.
+
+        UNVERIFIED FLOW: the collection table can't target a collid directly,
+        so this filters by game and opens the row's status link. With several
+        copies of one game the wrong row could open — acceptable only while
+        every 2018 entry is version-less; revisit after the first real run.
+        """
+        self._ensure_logged_in()
+        page = self._page
+        self._goto(
+            f"{BGG}/collection/user/{self._username}?objectid={job.bgg_id}&own=1"
+        )
+        row = (
+            page.get_by_role("row")
+            .filter(has_text=re.compile(re.escape(job.name), re.I))
+            .first
+        )
+        opener = row.get_by_role("link", name=re.compile("own", re.I)).first
+        dialog = self._open_dialog(opener)
+        if not self._select_version(dialog, job.version_name):
+            # Never guess a version: close without touching the entry.
+            dialog.get_by_role("button", name="Cancel").first.click()
+            raise RuntimeError(
+                f"version {job.version_name!r} not in picker — entry left untouched"
+            )
+        dialog.get_by_role("button", name="Save").click()
+        dialog.wait_for(state="hidden", timeout=15_000)
+        return "updated", "row-edit flow is unverified; confirm with --verify"
+
+
+def _process(
+    uploader: Uploader,
+    jobs: list[UploadJob],
+    log_path: Path,
+    *,
+    sleep: Callable[[float], None],
+    rng: random.Random,
+    now: Callable[[], str],
+) -> list[dict]:
+    results = []
+    for i, job in enumerate(jobs):
+        if i:
+            sleep(rng.uniform(2.0, 4.0))  # polite pacing between games (spec)
+        try:
+            if job.action == "add":
+                status, note = uploader.add_game(job)
+            else:
+                status, note = uploader.update_entry(job)
+        except Exception as exc:  # per-game isolation: log it, keep going
+            status, note = "failed", _scrub(f"{type(exc).__name__}: {exc}")
+        row = {
+            "action": job.action,
+            "bgg_id": job.bgg_id,
+            "collid": job.collid,
+            "name": job.name,
+            "version_id": job.version_id,
+            "status": status,
+            "timestamp": now(),
+            "error": note if status == "failed" else note,
+        }
+        append_log_row(log_path, row)
+        results.append(row)
+        suffix = f" — {note}" if note else ""
+        typer.echo(f"  {job.name}: {status}{suffix}")
+    return results
+
+
+def verify_uploads(log_rows: list[dict], collection: list[CollectionItem]) -> list[str]:
+    """Cross-check the log's successes against a fresh collection fetch."""
+    by_object: dict[int, list[CollectionItem]] = {}
+    by_collid: dict[int, CollectionItem] = {}
+    for item in collection:
+        by_object.setdefault(item.object_id, []).append(item)
+        by_collid[item.coll_id] = item
+
+    problems = []
+    latest: dict[tuple[str, str, str], dict] = {}
+    for row in log_rows:
+        latest[_row_key(row)] = row
+    for row in latest.values():
+        if row["status"] == "added":
+            copies = by_object.get(int(row["bgg_id"]), [])
+            if not copies:
+                problems.append(f"{row['name']}: logged added but not in collection")
+            elif row["version_id"] and not any(
+                str(c.version_id or "") == row["version_id"] for c in copies
+            ):
+                problems.append(
+                    f"{row['name']}: in collection but no copy has "
+                    f"version {row['version_id']}"
+                )
+        elif row["status"] == "updated":
+            item = by_collid.get(int(row["collid"]))
+            if item is None:
+                problems.append(
+                    f"{row['name']}: collid {row['collid']} not in collection"
+                )
+            elif str(item.version_id or "") != row["version_id"]:
+                problems.append(
+                    f"{row['name']}: collid {row['collid']} does not carry "
+                    f"version {row['version_id']}"
+                )
+    return problems
+
+
+def run_upload(
+    cfg: Config,
+    *,
+    dry_run: bool = False,
+    verify: bool = False,
+    retry_failed: bool = False,
+    limit: int | None = None,
+    headless: bool = False,
+    uploader: Uploader | None = None,
+    client: BGGClient | None = None,
+    storage_state: Path = STORAGE_STATE_PATH,
+    sleep: Callable[[float], None] = time.sleep,
+    rng: random.Random | None = None,
+    now: Callable[[], str] | None = None,
+) -> list[dict]:
+    rng = rng or random.Random()
+    now = now or (lambda: datetime.now(UTC).isoformat(timespec="seconds"))
+
+    marker = cfg.cache_dir / "STUB_FIXTURES.marker"
+    if marker.exists():
+        if not dry_run:
+            typer.echo(
+                f"Refusing to upload: {marker} exists — every resolved "
+                "version_id is a synthetic stub placeholder. Once "
+                "BGG_API_TOKEN arrives: delete both cache dirs, re-record "
+                "fixtures, `resolve --force`, re-review, re-diff."
+            )
+            raise typer.Exit(code=1)
+        typer.echo(
+            "WARNING: stub fixtures active — the ids below are SYNTHETIC "
+            "placeholders, not real BGG data.\n"
+        )
+
+    log_path = cfg.data_dir / "upload_log.csv"
+    to_add = _read_csv(cfg.data_dir / "to_add.csv")
+    to_update = _read_csv(cfg.data_dir / "to_update.csv")
+    log_rows = _read_csv(log_path)
+
+    jobs, skipped_done, skipped_failed = build_queue(
+        to_add, to_update, log_rows, retry_failed=retry_failed
+    )
+    if limit is not None:
+        jobs = jobs[:limit]
+
+    skip_note = ""
+    if skipped_done or skipped_failed:
+        skip_note = (
+            f" (skipping {skipped_done} already done, {skipped_failed} "
+            "previously failed — use --retry-failed)"
+        )
+
+    results: list[dict] = []
+    if dry_run:
+        typer.echo(f"Dry run: {len(jobs)} job(s) pending{skip_note}")
+        for job in jobs:
+            if job.action == "add":
+                version = f" [version {job.version_name}]" if job.version_name else ""
+                typer.echo(f"  would add {job.name} ({job.bgg_id}){version}")
+            else:
+                typer.echo(
+                    f"  would set version {job.version_name!r} on existing "
+                    f"entry collid {job.collid} ({job.name})"
+                )
+    elif jobs:
+        typer.echo(f"Uploading {len(jobs)} job(s){skip_note}")
+        if uploader is None:
+            if not (os.environ.get("BGG_USERNAME") and os.environ.get("BGG_PASSWORD")):
+                typer.echo(
+                    "BGG_USERNAME and BGG_PASSWORD env vars are required "
+                    "(direnv loads them from .env)."
+                )
+                raise typer.Exit(code=1)
+            with PlaywrightUploader(
+                cfg.bgg_username, storage_state=storage_state, headless=headless
+            ) as real:
+                results = _process(real, jobs, log_path, sleep=sleep, rng=rng, now=now)
+        else:
+            results = _process(uploader, jobs, log_path, sleep=sleep, rng=rng, now=now)
+        counts: dict[str, int] = {}
+        for row in results:
+            counts[row["status"]] = counts.get(row["status"], 0) + 1
+        summary = " · ".join(f"{n} {status}" for status, n in sorted(counts.items()))
+        typer.echo(f"\n{summary or 'nothing to do'}")
+    else:
+        typer.echo(f"Nothing to upload{skip_note}.")
+
+    if verify and not dry_run:
+        _run_verify(cfg, client, log_path)
+    return results
+
+
+def _run_verify(cfg: Config, client: BGGClient | None, log_path: Path) -> None:
+    if not cfg.bgg_username:
+        typer.echo("--verify needs bgg_username (config.toml or BGG_USERNAME).")
+        return
+    client = client or BGGClient(cfg.cache_dir, cfg.rate_limit_seconds)
+    try:
+        collection = client.collection_full(cfg.bgg_username, refresh=True)
+    except BGGAuthError as exc:
+        typer.echo(f"--verify skipped: {exc}")
+        return
+    problems = verify_uploads(_read_csv(log_path), collection)
+    if problems:
+        typer.echo("\nVerification problems:")
+        for line in problems:
+            typer.echo(f"  - {line}")
+    else:
+        typer.echo("\nVerification OK: every logged success is in the collection.")