From 376edd7819fdf6c81394023b96ad4ecd5b3ef2de Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Thu, 4 Jun 2026 06:34:34 -0700 Subject: [PATCH] feat: interactive review CLI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per sub-project-2 plan Task 12 / spec ยง8.2. Walks proposed records (fields + coords + rationale + best-effort ffmpeg preview), prompts accept/edit/skip/quit, and persists each approval via save_catalog rewrite. I/O seams (input_fn/now_fn/ out) and --no-preview make the walk testable hermetically; decision logic lives in the tested review core. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_review_cli.py | 103 ++++++++++++++++++++++++++ tools/review_cli.py | 154 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 257 insertions(+) create mode 100644 tests/test_review_cli.py create mode 100644 tools/review_cli.py diff --git a/tests/test_review_cli.py b/tests/test_review_cli.py new file mode 100644 index 0000000..0d50682 --- /dev/null +++ b/tests/test_review_cli.py @@ -0,0 +1,103 @@ +import io + +import pytest + +from hef.catalog import Record, load_catalog, save_catalog +from tools.review_cli import main + + +def make_record(**o): + base = dict( + id="a", + title="t", + source_url="u", + source_archive="nasa", + license="public_domain", + mode="video", + left=0, + right=0, + dark=0, + light=0, + duration_s=1, + file_path="nasa/a.mp4", + ) + base.update(o) + return Record(**base) + + +def test_help_parses(): + with pytest.raises(SystemExit) as e: + main(["--help"]) + assert e.value.code == 0 + + +def test_no_proposed_records(tmp_path): + catalog = tmp_path / "library.jsonl" + save_catalog([make_record(id="x", review_status="approved", reviewed_at="t")], catalog) + out = io.StringIO() + rc = main(["--catalog", str(catalog), "--no-preview"], out=out) + assert rc == 0 + assert "no proposed records" in out.getvalue() + + +def test_accept_flips_to_approved(tmp_path): + catalog = tmp_path / "library.jsonl" + save_catalog([make_record(id="a", review_status="proposed")], catalog) + out = io.StringIO() + rc = main( + ["--catalog", str(catalog), "--no-preview"], + input_fn=lambda prompt: "a", + now_fn=lambda: "2026-06-04T13:00:00+00:00", + out=out, + ) + assert rc == 0 + r = load_catalog(catalog)[0] + assert r.review_status == "approved" + assert r.reviewed_at == "2026-06-04T13:00:00+00:00" + + +def test_skip_leaves_proposed(tmp_path): + catalog = tmp_path / "library.jsonl" + save_catalog([make_record(id="a", review_status="proposed")], catalog) + rc = main( + ["--catalog", str(catalog), "--no-preview"], + input_fn=lambda prompt: "s", + out=io.StringIO(), + ) + assert rc == 0 + assert load_catalog(catalog)[0].review_status == "proposed" + + +def test_edit_overrides_coordinates(tmp_path): + catalog = tmp_path / "library.jsonl" + save_catalog([make_record(id="a", review_status="proposed")], catalog) + answers = iter(["e", "4", "1", "2", "3"]) + rc = main( + ["--catalog", str(catalog), "--no-preview"], + input_fn=lambda prompt: next(answers), + now_fn=lambda: "t", + out=io.StringIO(), + ) + assert rc == 0 + r = load_catalog(catalog)[0] + assert (r.left, r.right, r.dark, r.light) == (4, 1, 2, 3) + assert r.review_status == "approved" + + +def test_quit_stops_walk(tmp_path): + catalog = tmp_path / "library.jsonl" + save_catalog( + [ + make_record(id="a", review_status="proposed"), + make_record(id="b", review_status="proposed"), + ], + catalog, + ) + rc = main( + ["--catalog", str(catalog), "--no-preview"], + input_fn=lambda prompt: "q", + out=io.StringIO(), + ) + assert rc == 0 + statuses = {r.id: r.review_status for r in load_catalog(catalog)} + assert statuses == {"a": "proposed", "b": "proposed"} diff --git a/tools/review_cli.py b/tools/review_cli.py new file mode 100644 index 0000000..4fd2f71 --- /dev/null +++ b/tools/review_cli.py @@ -0,0 +1,154 @@ +"""Interactive review walk: `python -m tools.review_cli`. + +Thin shell over the tested review core (tools.review). Walks each proposed +record showing its mechanical fields, the proposed coordinates + rationale, and +a best-effort preview frame, then prompts accept/edit/skip/quit. Approvals are +persisted (full rewrite) after each one, so an interrupted session keeps its +progress. +""" + +from __future__ import annotations + +import argparse +import os +import platform +import shutil +import subprocess +import sys +import tempfile +from datetime import datetime, timezone +from pathlib import Path + +from hef.catalog import load_catalog, save_catalog, validate_catalog +from tools.mediatools import extract_frame +from tools.review import approve, proposed_records + +DEFAULT_CATALOG = "catalog/library.jsonl" +DEFAULT_MEDIA_ROOT = "./media" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="tools.review_cli", + description="Walk proposed records and approve/correct their coordinates.", + ) + p.add_argument("--catalog", default=DEFAULT_CATALOG) + p.add_argument( + "--media-root", + default=os.environ.get("HEF_MEDIA_ROOT", DEFAULT_MEDIA_ROOT), + ) + p.add_argument("--no-preview", action="store_true", help="skip preview rendering") + return p + + +def _show(rec, out) -> None: + print(f"\n{'=' * 60}", file=out) + print(f" id {rec.id}", file=out) + print(f" title {rec.title}", file=out) + print(f" source {rec.source_archive} {rec.source_url}", file=out) + attr = f" ({rec.attribution})" if rec.attribution else "" + print(f" license {rec.license}{attr}", file=out) + print(f" media mode={rec.mode} {rec.duration_s}s {rec.resolution}", file=out) + if rec.dominant_color: + print(f" color {rec.dominant_color}", file=out) + print( + f" proposed left={rec.left} right={rec.right} dark={rec.dark} light={rec.light}", + file=out, + ) + print(f" rationale {rec.rationale}", file=out) + if rec.notes: + print(f" notes {rec.notes}", file=out) + + +def _open_cmd(): + return "open" if platform.system() == "Darwin" else "xdg-open" + + +def _preview(rec, media_root, out) -> None: + """Best-effort preview; never fatal. Requires ffmpeg for video frames.""" + if not shutil.which("ffmpeg"): + return + src = Path(media_root) / rec.file_path + if not src.exists(): + print(f" (preview: media not found at {src})", file=out) + return + try: + midpoint = rec.duration_s / 2 if rec.duration_s else 0.0 + if rec.mode in ("video", "av"): + dest = Path(tempfile.gettempdir()) / f"hef-preview-{rec.id}.png" + extract_frame(src, dest, midpoint_s=midpoint) + else: # audio -> waveform thumbnail + dest = Path(tempfile.gettempdir()) / f"hef-preview-{rec.id}.png" + subprocess.run( + [ + "ffmpeg", "-v", "quiet", "-y", "-i", str(src), + "-filter_complex", "showwavespic=s=640x240", "-frames:v", "1", + str(dest), + ], + check=True, + ) + opener = shutil.which(_open_cmd()) + if opener: + subprocess.run([opener, str(dest)], check=False) + except Exception as exc: # noqa: BLE001 - preview is best-effort + print(f" (preview failed: {exc})", file=out) + + +def _prompt_coords(rec, input_fn, out): + from hef.selection import Coordinate + + def _ask(axis, current): + raw = input_fn(f" {axis} [{current}]: ").strip() + if not raw: + return current + try: + return max(0, min(4, int(raw))) + except ValueError: + print(f" (not an int, keeping {current})", file=out) + return current + + return Coordinate( + _ask("left", rec.left), + _ask("right", rec.right), + _ask("dark", rec.dark), + _ask("light", rec.light), + ) + + +def main(argv=None, *, input_fn=input, now_fn=_now, out=sys.stdout) -> int: + args = _build_parser().parse_args(argv) + + records = load_catalog(args.catalog) + validate_catalog(records) + pending = proposed_records(records) + if not pending: + print("no proposed records to review", file=out) + return 0 + + pos = {r.id: i for i, r in enumerate(records)} + print(f"{len(pending)} proposed record(s) to review", file=out) + + for rec in pending: + _show(rec, out) + if not args.no_preview: + _preview(rec, args.media_root, out) + choice = input_fn("[a]ccept / [e]dit / [s]kip / [q]uit > ").strip().lower() + if choice == "q": + break + if choice == "s" or choice not in ("a", "e"): + continue + coordinate = _prompt_coords(rec, input_fn, out) if choice == "e" else None + approved = approve(records[pos[rec.id]], reviewed_at=now_fn(), coordinate=coordinate) + records[pos[rec.id]] = approved + save_catalog(records, args.catalog) + print(f" approved {rec.id}", file=out) + + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main())