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"}

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())