human-experience-filter-art/tools/review_cli.py

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