#!/usr/bin/env python3 """Build a lightweight local feedback proxy bound to an exact-30fps master. The accepted 1080x1920 master is never changed. A verified 360x640 CFR proxy is published atomically inside ``.feedback/`` for fast browser review; the manifest binds its frames and checksum back to the accepted master. """ from __future__ import annotations import argparse import hashlib import json import os import subprocess import tempfile from fractions import Fraction from pathlib import Path PROXY_NAME = "review-proxy.mp4" PROXY_WIDTH = 360 PROXY_HEIGHT = 640 HTML = r''' Frame feedback
Frame feedback360×640 review proxy · exact 30fps frame clock
0:00.000 · frame 0Loading review proxy…

Keys: Space play/pause · ←/→ ±1 frame · Shift+←/→ ±10 frames. This smaller proxy maps every comment to the same exact frame in the accepted master.

''' SERVER = '''#!/usr/bin/env python3 """Serve this feedback artifact only to this Mac, on a loopback port.""" from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import quote import argparse import webbrowser class LoopbackOnlyHandler(SimpleHTTPRequestHandler): def parse_request(self): if not super().parse_request(): return False host = self.headers.get("Host", "").rsplit(":", 1)[0].lower() if host not in ("127.0.0.1", "localhost"): self.send_error(421, "Misdirected Request") return False return True artifact = Path(__file__).resolve().parent root = artifact.parent parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--no-open", action="store_true") parser.add_argument("--port", type=int, default=0, help="stable loopback port; 0 chooses a free port (default)") args = parser.parse_args() if not 0 <= args.port <= 65535: parser.error("--port must be between 0 and 65535") handler = partial(LoopbackOnlyHandler, directory=str(root)) try: server = ThreadingHTTPServer(("127.0.0.1", args.port), handler) except OSError as exc: parser.error(f"could not bind loopback port {args.port}: {exc}") with server: url = f"http://127.0.0.1:{server.server_port}/{quote(artifact.name)}/index.html" print(f"Local-only proxy review: {url}", flush=True) if not args.no_open: webbrowser.open(url) try: server.serve_forever() except KeyboardInterrupt: pass ''' def _json_from(argv: list[str]) -> dict: result = subprocess.run(argv, capture_output=True, text=True, check=False) if result.returncode: raise ValueError(f"ffprobe failed: {result.stderr.strip() or result.returncode}") try: return json.loads(result.stdout) except json.JSONDecodeError as exc: raise ValueError("ffprobe returned invalid JSON") from exc def probe_video(video: Path, ffprobe: str, *, width: int, height: int, label: str) -> dict: stream_data = _json_from([ ffprobe, "-v", "error", "-show_streams", "-of", "json", str(video), ]) streams = [stream for stream in stream_data.get("streams", []) if stream.get("codec_type") == "video"] if len(streams) != 1: raise ValueError(f"{label} must contain exactly one video stream") stream = streams[0] if (stream.get("width"), stream.get("height")) != (width, height): raise ValueError(f"{label} must be exact {width}x{height}") try: rates = {key: Fraction(str(stream[key])) for key in ("avg_frame_rate", "r_frame_rate")} except (KeyError, ValueError, ZeroDivisionError) as exc: raise ValueError("ffprobe did not provide valid frame-rate fields") from exc if rates != {"avg_frame_rate": Fraction(30), "r_frame_rate": Fraction(30)}: raise ValueError(f"{label} must be exact CFR 30fps (avg_frame_rate and r_frame_rate must both equal 30/1)") frame_data = _json_from([ ffprobe, "-v", "error", "-select_streams", "v:0", "-show_frames", "-show_entries", "frame=best_effort_timestamp_time", "-of", "json", str(video), ]) times = [float(frame["best_effort_timestamp_time"]) for frame in frame_data.get("frames", []) if "best_effort_timestamp_time" in frame] if len(times) < 2 or any(abs((right - left) - (1 / 30)) > 1e-6 for left, right in zip(times, times[1:])): raise ValueError(f"{label} must be exact CFR 30fps (decoded frame timestamps are not 1/30 second apart)") return { "cfr_30fps": True, "width": width, "height": height, "avg_frame_rate": stream["avg_frame_rate"], "r_frame_rate": stream["r_frame_rate"], "verified_frame_timestamp_count": len(times), } def probe_master(master: Path, ffprobe: str) -> dict: return probe_video(master, ffprobe, width=1080, height=1920, label="master") def encode_proxy(master: Path, output: Path, ffmpeg: str, ffprobe: str, master_probe: dict) -> tuple[Path, dict]: proxy = output / PROXY_NAME descriptor, temporary_name = tempfile.mkstemp( prefix=".review-proxy-", suffix=".mp4", dir=output, ) os.close(descriptor) temporary = Path(temporary_name) try: result = subprocess.run([ ffmpeg, "-hide_banner", "-loglevel", "error", "-y", "-i", str(master), "-map", "0:v:0", "-map", "0:a:0?", "-vf", f"scale={PROXY_WIDTH}:{PROXY_HEIGHT}:flags=lanczos", "-fps_mode", "cfr", "-r", "30", "-c:v", "libx264", "-preset", "veryfast", "-crf", "20", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "96k", "-movflags", "+faststart", str(temporary), ], capture_output=True, text=True, check=False) if result.returncode: raise ValueError(f"ffmpeg proxy encode failed: {result.stderr.strip() or result.returncode}") proxy_probe = probe_video( temporary, ffprobe, width=PROXY_WIDTH, height=PROXY_HEIGHT, label="review proxy", ) if proxy_probe["verified_frame_timestamp_count"] != master_probe["verified_frame_timestamp_count"]: raise ValueError( "review proxy frame count does not match master " f"({proxy_probe['verified_frame_timestamp_count']} != " f"{master_probe['verified_frame_timestamp_count']})" ) temporary.replace(proxy) return proxy, proxy_probe finally: temporary.unlink(missing_ok=True) def build(master: Path, output: Path, *, ffprobe: str = "ffprobe", ffmpeg: str = "ffmpeg") -> Path: master = master.resolve() output = output.resolve() if not master.is_file(): raise ValueError(f"master does not exist: {master}") if output.parent != master.parent: raise ValueError("artifact must be a sibling directory of its master so it binds ../") if output.exists() and not output.is_dir(): raise ValueError(f"artifact target is not a directory: {output}") probe = probe_master(master, ffprobe) output.mkdir(parents=True, exist_ok=True) proxy, proxy_probe = encode_proxy(master, output, ffmpeg, ffprobe, probe) master_sha256 = hashlib.sha256(master.read_bytes()).hexdigest() page = HTML.replace("__VIDEO__", PROXY_NAME).replace( "__STORAGE_KEY__", json.dumps(f"frame-feedback:{master_sha256}"), ) (output / "index.html").write_text(page, encoding="utf-8") manifest = { "master": master.name, "master_sha256": master_sha256, "video_src": PROXY_NAME, "proxy_sha256": hashlib.sha256(proxy.read_bytes()).hexdigest(), "fps": 30, "resolution": [PROXY_WIDTH, PROXY_HEIGHT], "offline": True, "probe": probe, "proxy_probe": proxy_probe, } (output / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") (output / "serve.py").write_text(SERVER, encoding="utf-8") return output def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("master", type=Path) parser.add_argument("--out", type=Path, help="default: .feedback beside master") parser.add_argument("--ffprobe", default="ffprobe", help="ffprobe executable (default: ffprobe)") parser.add_argument("--ffmpeg", default="ffmpeg", help="ffmpeg executable (default: ffmpeg)") args = parser.parse_args() output = args.out or args.master.with_name(args.master.name + ".feedback") try: artifact = build(args.master, output, ffprobe=args.ffprobe, ffmpeg=args.ffmpeg) except ValueError as exc: parser.error(str(exc)) print(f"bound lightweight review proxy: {args.master.resolve()} -> {artifact / 'index.html'}") if __name__ == "__main__": main()