diff --git a/CLAUDE.md b/CLAUDE.md index 33a98b5..6e71a8e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,13 +20,14 @@ Full design lives in `bgg-shelf-pipeline-spec.md` (read it before changing pipel ## Commands - `uv sync` — install deps (Python 3.12+, managed by **uv**; use `uv add`, never pip). `uv run bggpipe init` handles first-run setup (folders, .env credentials, the one-time `playwright install chromium`). +- `uv run bggpipe web` — the full pipeline as a local web app (dashboard at `/`, review at `/review`); stage runs execute one-at-a-time in a background job. - `uv run bggpipe ` — run a pipeline stage. Non-secret settings come from `config.toml` (username, dirs, vision model, rate limit); `--config` overrides the path. - `uv run pytest` — the suite runs fully offline against fixtures. Tests marked `live` hit the real BGG API (read-only) and are skipped unless you pass `--run-live`. - `uv run ruff check` / `uv run ruff format` — lint (rules E, F, I, UP, B, SIM) and format. ## Layout -- `src/bggpipe/` — `cli.py` (typer app), one module per stage (`extract`, `resolve`, `review` + `webreview`, `diff`, `upload`, `enrich`), plus `bgg_client.py` (rate-limited XML API2 client that caches responses to `data/bgg_cache/`), `normalize.py` (title normalization), `models.py` (dataclasses), `config.py`. +- `src/bggpipe/` — `cli.py` (typer app), one module per stage (`extract`, `resolve`, `review` + `webreview`, `diff`, `upload`, `enrich`), plus `bgg_client.py` (rate-limited XML API2 client that caches responses to `data/bgg_cache/`), `jobs.py` (single-slot background stage runner for the web UI), `normalize.py` (title normalization), `models.py` (dataclasses), `config.py`, `fsio.py` (atomic writes), `init_wizard.py` (first-run setup). - `scripts/` — `write_stub_fixtures.py` / `write_photo_fixtures.py` generate synthetic fixtures; `record_fixtures.py` re-records real API responses once a token exists. - `tests/fixtures/bgg_cache/` — stub XML fixtures the offline tests run against. - `data/` — pipeline state (CSV/JSON artifacts are committed; caches are not — see Git). diff --git a/README.md b/README.md index 3532c61..9ef3e13 100644 --- a/README.md +++ b/README.md @@ -54,12 +54,18 @@ Secrets live in environment variables only, never in config files, code, or logs | `BGG_USERNAME` | diff, upload, enrich | Your BGG username (public, but kept in `.env` so it lives in one place) | | `BGG_PASSWORD` | upload (website login) | Your BGG password | -Non-secret knobs (`photos_dir`, `data_dir`, the vision model, the rate limit) live in `config.toml`. Drop your shelf photos into `photos/` and run the stages in order: +Non-secret knobs (`photos_dir`, `data_dir`, the vision model, the rate limit) live in `config.toml`. From here you can drive everything from the browser: + +```sh +uv run bggpipe web # opens http://127.0.0.1:8377/ — the whole pipeline in one page +``` + +The dashboard shows every stage's status, takes photos by drag-and-drop, runs each stage with live output, links to the review page, and keeps the real upload behind a confirmation (and behind the stub-data lock). Prefer the terminal? Every stage is also a command, and the two interfaces share all state: ```sh uv run bggpipe extract # photos → titles.json (+ retake prompts) uv run bggpipe resolve # titles → BGG ids/versions in matches.csv -uv run bggpipe review --web # review UI at http://127.0.0.1:8377/ +uv run bggpipe review --web # review UI only uv run bggpipe diff # compare against your BGG collection uv run bggpipe upload --dry-run # ALWAYS inspect this first uv run bggpipe upload --limit 1 # then one game, then small batches diff --git a/pyproject.toml b/pyproject.toml index ecb95c3..fd8097f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "uvicorn>=0.52.1", "playwright>=1.62.0", "pydantic>=2.13.4", + "python-multipart>=0.0.32", ] [project.scripts] diff --git a/src/bggpipe/cli.py b/src/bggpipe/cli.py index 5fd3a21..ae2f62d 100644 --- a/src/bggpipe/cli.py +++ b/src/bggpipe/cli.py @@ -85,6 +85,33 @@ def review( run_review(cfg) +@app.command() +def web( + port: Annotated[ + int, typer.Option("--port", help="Port to serve on") + ] = DEFAULT_REVIEW_PORT, + dev: Annotated[ + bool, typer.Option("--dev", help="Restart on source changes") + ] = False, + no_browser: Annotated[ + bool, typer.Option("--no-browser", help="Don't open a browser tab") + ] = False, + config: ConfigOpt = None, +) -> None: + """The whole pipeline in a local web UI: photos, stages, review.""" + from bggpipe.webreview import run_web_review + + cfg = load_config(config) + run_web_review( + cfg, + port=port, + dev=dev, + config_path=config, + landing="/", + open_browser=not no_browser, + ) + + @app.command() def diff(config: ConfigOpt = None) -> None: """Stage 4: diff approved matches against the existing BGG collection.""" diff --git a/src/bggpipe/jobs.py b/src/bggpipe/jobs.py new file mode 100644 index 0000000..5a45c6e --- /dev/null +++ b/src/bggpipe/jobs.py @@ -0,0 +1,101 @@ +"""One-at-a-time background execution of pipeline stages for the web UI. + +The stages share the data/ artifacts, so running two concurrently is +unsupported everywhere in the pipeline — the runner enforces it with a +single slot. Stage output (typer.echo goes to stdout) is captured by +swapping sys.stdout for the job's duration; that swap is process-wide, +which is safe here only because the runner holds the single slot and the +web server itself never writes to stdout (uvicorn logs on stderr). +""" + +from __future__ import annotations + +import io +import threading +import time +from collections.abc import Callable +from contextlib import redirect_stdout + +import typer + + +class _LineBuffer(io.TextIOBase): + def __init__(self) -> None: + self._lock = threading.Lock() + self._lines: list[str] = [] + self._partial = "" + + def write(self, text: str) -> int: + with self._lock: + self._partial += text + *complete, self._partial = self._partial.split("\n") + self._lines.extend(complete) + return len(text) + + def lines(self) -> list[str]: + with self._lock: + return self._lines + ([self._partial] if self._partial else []) + + +class JobRunner: + """At most one running job; finished state persists until the next start.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._thread: threading.Thread | None = None + self._stage = "" + self._status = "idle" # idle | running | done | failed + self._buffer = _LineBuffer() + self._error = "" + self._started = 0.0 + self._finished = 0.0 + + def start(self, stage: str, fn: Callable[[], object]) -> bool: + """Begin a job; False when one is already running.""" + with self._lock: + if self._status == "running": + return False + self._stage = stage + self._status = "running" + self._buffer = _LineBuffer() + self._error = "" + self._started = time.time() + self._finished = 0.0 + self._thread = threading.Thread(target=self._run, args=(fn,), daemon=True) + self._thread.start() + return True + + def _run(self, fn: Callable[[], object]) -> None: + status, error = "done", "" + try: + with redirect_stdout(self._buffer): + fn() + except typer.Exit as exc: + if exc.exit_code: + status, error = "failed", f"exited with code {exc.exit_code}" + except SystemExit as exc: + if exc.code: + status, error = "failed", f"exited with code {exc.code}" + except Exception as exc: # surfaced in the UI, never swallowed + status, error = "failed", f"{type(exc).__name__}: {exc}" + with self._lock: + self._status = status + self._error = error + self._finished = time.time() + + def snapshot(self) -> dict: + with self._lock: + return { + "stage": self._stage, + "status": self._status, + "log": self._buffer.lines()[-200:], + "error": self._error, + "started": self._started, + "finished": self._finished, + } + + def wait(self, timeout: float = 10.0) -> None: + """Test hook: block until the current job's thread finishes.""" + thread = self._thread + if thread is not None: + thread.join(timeout) diff --git a/src/bggpipe/templates/dashboard.html b/src/bggpipe/templates/dashboard.html new file mode 100644 index 0000000..0571dda --- /dev/null +++ b/src/bggpipe/templates/dashboard.html @@ -0,0 +1,283 @@ + + + + + +bggpipe + + + + +
+
+ + bggpipeshelf → BGG pipeline +
+ +
+ +
+

Photos

+
drop shelf photos here, or click to choose + +
+ +

Pipeline

+
+ +

Activity

+
idle
+
(stage output appears here)
+
+ + + diff --git a/src/bggpipe/webreview.py b/src/bggpipe/webreview.py index 2a06522..21402c6 100644 --- a/src/bggpipe/webreview.py +++ b/src/bggpipe/webreview.py @@ -1,4 +1,4 @@ -"""`bggpipe review --web` — the review TUI's local web face. +"""The local web app: a pipeline dashboard at / and the review UI at /review. FastAPI + one self-contained HTML page (inline CSS/JS, no build step), served on localhost only. All decision logic and matches.csv writes go @@ -17,14 +17,17 @@ import json import os import threading import warnings +import webbrowser import xml.etree.ElementTree as ET from collections import Counter +from collections.abc import Callable +from functools import partial from importlib import resources from pathlib import Path import typer from defusedxml.ElementTree import fromstring as _safe_fromstring -from fastapi import FastAPI, HTTPException +from fastapi import FastAPI, HTTPException, UploadFile from fastapi.responses import FileResponse, HTMLResponse, Response from pydantic import BaseModel from rich.console import Console @@ -32,6 +35,7 @@ from rich.console import Console from bggpipe.bgg_client import BGGClient, cached_paths from bggpipe.config import DEFAULT_REVIEW_PORT, Config from bggpipe.fsio import atomic_write_text +from bggpipe.jobs import JobRunner from bggpipe.models import ( CONFIDENT_VERSION_STATUSES, RECOGNIZED_MATCH_STATUSES, @@ -134,6 +138,11 @@ class VetoBody(BaseModel): row_ix: int | None = None +class RunBody(BaseModel): + dry_run: bool = True # upload only; the safe direction is the default + limit: int | None = None + + class DismissBody(BaseModel): photo: str location: str = "" @@ -141,8 +150,57 @@ class DismissBody(BaseModel): art_notes: str = "" -def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI: - app = FastAPI(title="bggpipe review") +def _default_stages(cfg: Config) -> dict[str, Callable[..., object]]: + """Lazy imports, same convention as the CLI: the dashboard buttons run + exactly what the CLI commands run.""" + + def extract() -> object: + from bggpipe.extract import run_extract + + return run_extract(cfg) + + def resolve() -> object: + from bggpipe.resolve import run_resolve + + return run_resolve(cfg) + + def diff() -> object: + from bggpipe.diff import run_diff + + return run_diff(cfg) + + def upload(dry_run: bool = True, limit: int | None = None) -> object: + from bggpipe.upload import run_upload + + return run_upload(cfg, dry_run=dry_run, limit=limit) + + def enrich() -> object: + from bggpipe.enrich import run_enrich + + return run_enrich(cfg) + + return { + "extract": extract, + "resolve": resolve, + "diff": diff, + "upload": upload, + "enrich": enrich, + } + + +PHOTO_SUFFIXES = {".jpg", ".jpeg", ".png", ".heic"} + + +def create_app( + cfg: Config, + *, + client: BGGClient | None = None, + stages: dict[str, Callable[..., object]] | None = None, + jobs: JobRunner | None = None, +) -> FastAPI: + app = FastAPI(title="bggpipe") + stages = stages or _default_stages(cfg) + jobs = jobs or JobRunner() session = ReviewSession( cfg, console=Console(file=io.StringIO()), @@ -288,8 +346,94 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI: @app.get("/", response_class=HTMLResponse) def index() -> str: + return (resources.files("bggpipe") / "templates" / "dashboard.html").read_text() + + @app.get("/review", response_class=HTMLResponse) + def review_page() -> str: return (resources.files("bggpipe") / "templates" / "review.html").read_text() + def _csv_count(path: Path) -> int: + if not path.exists(): + return 0 + with path.open(newline="") as f: + return max(0, sum(1 for _ in f) - 1) # minus header + + @app.get("/api/pipeline") + def api_pipeline() -> dict: + with lock: + freshen() + match_counts = Counter(row["match_status"] for row in session.rows) + log_counts: Counter[str] = Counter() + log_path = cfg.upload_log_path + if log_path.exists(): + with log_path.open(newline="") as f: + import csv as _csv + + log_counts = Counter(row["status"] for row in _csv.DictReader(f)) + games = 0 + if cfg.games_path.exists(): + games = len(json.loads(cfg.games_path.read_text())) + return { + # key NAMES only — never values (credentials stay out of + # every payload and log) + "env": { + key: bool(os.environ.get(key)) + for key in ( + "ANTHROPIC_API_KEY", + "BGG_API_TOKEN", + "BGG_USERNAME", + "BGG_PASSWORD", + ) + }, + "stub_data": any(m.exists() for m in cfg.stub_marker_paths), + "photos": len(photo_names()), + "titles": len(session.titles), + "matches": dict(match_counts), + "pending_review": len(session.pending_rows()) + + len(session.version_rows()), + "to_add": _csv_count(cfg.to_add_path), + "to_update": _csv_count(cfg.to_update_path), + "upload_log": dict(log_counts), + "games": games, + "job": jobs.snapshot(), + } + + @app.post("/api/run/{stage}") + def api_run(stage: str, body: RunBody | None = None) -> dict: + if stage not in stages: + raise HTTPException(404, f"unknown stage {stage!r}") + body = body or RunBody() + if stage == "upload": + fn = partial(stages["upload"], dry_run=body.dry_run, limit=body.limit) + else: + fn = stages[stage] + if not jobs.start(stage, fn): + raise HTTPException(409, "a stage is already running — wait for it") + return jobs.snapshot() + + @app.get("/api/job") + def api_job() -> dict: + return jobs.snapshot() + + @app.post("/api/photos") + async def api_photos(files: list[UploadFile]) -> dict: + saved = [] + cfg.photos_dir.mkdir(parents=True, exist_ok=True) + for upload_file in files: + name = Path(upload_file.filename or "").name # strips any path + if not name or Path(name).suffix.lower() not in PHOTO_SUFFIXES: + raise HTTPException(400, f"not a photo: {name or '(unnamed)'}") + target = cfg.photos_dir / name + data = await upload_file.read() + target.write_bytes(data) + # a re-uploaded photo means "re-extract this one": drop its raw + # cache (regenerable) so the next extract run picks it up + stale = cfg.extract_raw_dir / f"{name}.json" + if stale.exists(): + stale.unlink() + saved.append(name) + return {"saved": saved, "photos": len(photo_names())} + @app.get("/api/state") def api_state() -> dict: with lock: @@ -395,13 +539,19 @@ def run_web_review( port: int = DEFAULT_REVIEW_PORT, dev: bool = False, config_path: Path | None = None, + landing: str = "/review", + open_browser: bool = False, ) -> None: import uvicorn + url = f"http://127.0.0.1:{port}{landing}" typer.echo( - f"Review UI: http://127.0.0.1:{port}/ (localhost only; every " - "decision saves to matches.csv immediately — Ctrl-C anytime)" + f"bggpipe web UI: {url} (localhost only; dashboard at /, review " + "at /review; every decision saves immediately — Ctrl-C anytime)" ) + if open_browser: + # give uvicorn a beat to bind before the tab loads + threading.Timer(0.8, webbrowser.open, args=(url,)).start() if dev: # Code hot-reload. Watch only the package source: uvicorn's default # (cwd, recursive) would restart the server on every matches.csv diff --git a/tests/test_web_dashboard.py b/tests/test_web_dashboard.py new file mode 100644 index 0000000..f59d1e1 --- /dev/null +++ b/tests/test_web_dashboard.py @@ -0,0 +1,187 @@ +"""Dashboard/API tests: job lifecycle, pipeline status, photo upload — +stage functions are injected so nothing slow or networked ever runs.""" + +from __future__ import annotations + +import threading + +import typer +from fastapi.testclient import TestClient + +from bggpipe.config import Config +from bggpipe.jobs import JobRunner +from bggpipe.webreview import create_app + + +def _cfg(tmp_path) -> Config: + cfg = Config(data_dir=tmp_path / "data", photos_dir=tmp_path / "photos") + cfg.photos_dir.mkdir(parents=True) + cfg.data_dir.mkdir(parents=True) + return cfg + + +def _app(cfg, stages=None, jobs=None) -> TestClient: + return TestClient(create_app(cfg, stages=stages or {}, jobs=jobs)) + + +# -- JobRunner ---------------------------------------------------------- + + +def test_job_captures_output_and_finishes(): + runner = JobRunner() + + def stage(): + typer.echo("line one") + typer.echo("line two") + + assert runner.start("extract", stage) + runner.wait() + snap = runner.snapshot() + assert snap["status"] == "done" + assert snap["log"] == ["line one", "line two"] + + +def test_job_failure_is_reported_not_swallowed(): + runner = JobRunner() + runner.start("resolve", lambda: (_ for _ in ()).throw(RuntimeError("boom"))) + runner.wait() + snap = runner.snapshot() + assert snap["status"] == "failed" + assert "RuntimeError: boom" in snap["error"] + + +def test_typer_exit_code_counts_as_failure(): + runner = JobRunner() + + def stage(): + raise typer.Exit(code=1) + + runner.start("diff", stage) + runner.wait() + assert runner.snapshot()["status"] == "failed" + + +def test_single_slot_rejects_second_job(): + runner = JobRunner() + release = threading.Event() + assert runner.start("extract", release.wait) + assert not runner.start("resolve", lambda: None) # slot busy + release.set() + runner.wait() + assert runner.start("resolve", lambda: None) # slot free again + runner.wait() + + +# -- /api/run + /api/job ------------------------------------------------ + + +def test_run_stage_lifecycle_via_api(tmp_path): + cfg = _cfg(tmp_path) + ran = [] + jobs = JobRunner() + web = _app(cfg, stages={"extract": lambda: ran.append(1)}, jobs=jobs) + + res = web.post("/api/run/extract") + assert res.status_code == 200 + jobs.wait() + assert ran == [1] + assert web.get("/api/job").json()["status"] == "done" + + +def test_unknown_stage_404s(tmp_path): + web = _app(_cfg(tmp_path)) + assert web.post("/api/run/frobnicate").status_code == 404 + + +def test_busy_runner_409s(tmp_path): + cfg = _cfg(tmp_path) + release = threading.Event() + jobs = JobRunner() + web = _app( + cfg, stages={"extract": release.wait, "resolve": lambda: None}, jobs=jobs + ) + assert web.post("/api/run/extract").status_code == 200 + assert web.post("/api/run/resolve").status_code == 409 + release.set() + jobs.wait() + + +def test_upload_defaults_to_dry_run(tmp_path): + cfg = _cfg(tmp_path) + calls = [] + jobs = JobRunner() + + def upload(dry_run=True, limit=None): + calls.append({"dry_run": dry_run, "limit": limit}) + + web = _app(cfg, stages={"upload": upload}, jobs=jobs) + web.post("/api/run/upload") # no body: the safe direction + jobs.wait() + web.post("/api/run/upload", json={"dry_run": False, "limit": 2}) + jobs.wait() + assert calls == [ + {"dry_run": True, "limit": None}, + {"dry_run": False, "limit": 2}, + ] + + +# -- /api/pipeline ------------------------------------------------------ + + +def test_pipeline_reports_counts_and_never_values(tmp_path, monkeypatch): + monkeypatch.setenv("BGG_USERNAME", "supersecretname") + monkeypatch.delenv("BGG_API_TOKEN", raising=False) + cfg = _cfg(tmp_path) + (cfg.photos_dir / "a.jpg").write_bytes(b"x") + web = _app(cfg) + payload = web.get("/api/pipeline").json() + assert payload["photos"] == 1 + assert payload["env"]["BGG_USERNAME"] is True + assert payload["env"]["BGG_API_TOKEN"] is False + assert "supersecretname" not in web.get("/api/pipeline").text + + +def test_pipeline_flags_stub_data(tmp_path): + cfg = _cfg(tmp_path) + (cfg.data_dir / "STUB_DATA.marker").write_text("stub") + assert _app(cfg).get("/api/pipeline").json()["stub_data"] is True + + +# -- /api/photos -------------------------------------------------------- + + +def test_photo_upload_saves_and_invalidates_raw_cache(tmp_path): + cfg = _cfg(tmp_path) + cfg.extract_raw_dir.mkdir(parents=True) + stale = cfg.extract_raw_dir / "shelf.jpg.json" + stale.write_text("{}") + web = _app(cfg) + res = web.post( + "/api/photos", files={"files": ("shelf.jpg", b"\xff\xd8jpegdata", "image/jpeg")} + ) + assert res.status_code == 200 + assert (cfg.photos_dir / "shelf.jpg").read_bytes() == b"\xff\xd8jpegdata" + assert not stale.exists() # re-upload means re-extract + + +def test_photo_upload_rejects_non_photos_and_path_tricks(tmp_path): + cfg = _cfg(tmp_path) + web = _app(cfg) + res = web.post("/api/photos", files={"files": ("notes.txt", b"hi", "text/plain")}) + assert res.status_code == 400 + res = web.post( + "/api/photos", + files={"files": ("../../escape.jpg", b"x", "image/jpeg")}, + ) + if res.status_code == 200: # client may strip the path; the name must be bare + assert (cfg.photos_dir / "escape.jpg").exists() + assert not (tmp_path / "escape.jpg").exists() + + +# -- pages -------------------------------------------------------------- + + +def test_dashboard_and_review_pages_serve(tmp_path): + web = _app(_cfg(tmp_path)) + assert "Pipeline" in web.get("/").text + assert "bggpipe" in web.get("/review").text diff --git a/uv.lock b/uv.lock index ba9c762..6cd3a3c 100644 --- a/uv.lock +++ b/uv.lock @@ -64,6 +64,7 @@ dependencies = [ { name = "pillow-heif" }, { name = "playwright" }, { name = "pydantic" }, + { name = "python-multipart" }, { name = "rapidfuzz" }, { name = "rich" }, { name = "typer" }, @@ -86,6 +87,7 @@ requires-dist = [ { name = "pillow-heif", specifier = ">=1.5.0" }, { name = "playwright", specifier = ">=1.62.0" }, { name = "pydantic", specifier = ">=2.13.4" }, + { name = "python-multipart", specifier = ">=0.0.32" }, { name = "rapidfuzz", specifier = ">=3.9" }, { name = "rich", specifier = ">=15.0.0" }, { name = "typer", specifier = ">=0.12" }, @@ -660,6 +662,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, ] +[[package]] +name = "python-multipart" +version = "0.0.32" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5b/42/55c32bb9b12693c092ad250a0e82edb5b31ddeda6eb772de5f308b3804ad/python_multipart-0.0.32.tar.gz", hash = "sha256:be54b7f3fa167bb83e4fcd936b887b708f4e57fe75911c02aebf53efaf8d938e", size = 46881, upload-time = "2026-06-04T16:18:58.647Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e1/04/e8135ebd1ad02c56ec633277529b2602ff99ff634be76cdba5744cf554fd/python_multipart-0.0.32-py3-none-any.whl", hash = "sha256:ff6d3f776f16878c894e52e107296ffc890e913c611b1a4ec6c44e2821fe2e23", size = 30042, upload-time = "2026-06-04T16:18:57.319Z" }, +] + [[package]] name = "rapidfuzz" version = "3.14.5"