ai-presenter-video

1#!/usr/bin/env python32"""Validate a presenter-video job and write a machine-readable preflight report."""3 4from __future__ import annotations5 6import json7import shutil8import subprocess9import sys10from pathlib import Path11from typing import Any12 13 14def probe(path: Path) -> dict[str, Any]:15    result = subprocess.run(16        [17            "ffprobe",18            "-v",19            "error",20            "-show_entries",21            "stream=index,codec_type,codec_name,width,height,pix_fmt,sample_rate,channels,r_frame_rate:format=duration,size,bit_rate",22            "-of",23            "json",24            str(path),25        ],26        check=True,27        capture_output=True,28        text=True,29    )30    return json.loads(result.stdout)31 32 33def require_file(value: str, label: str, errors: list[str]) -> Path | None:34    if not value:35        errors.append(f"missing {label}")36        return None37    path = Path(value).expanduser().resolve()38    if not path.is_file():39        errors.append(f"{label} is not a file: {path}")40        return None41    return path42 43 44def write_atomic(path: Path, text: str) -> None:45    temporary = path.with_suffix(path.suffix + ".tmp")46    temporary.write_text(text, encoding="utf-8")47    temporary.replace(path)48 49 50def main() -> int:51    if len(sys.argv) != 2:52        print("Usage: preflight.py ~/Videos/my-presenter-video/job.json", file=sys.stderr)53        return 6454    if not shutil.which("ffprobe"):55        print("ERROR: ffprobe is required", file=sys.stderr)56        return 257 58    job_path = Path(sys.argv[1]).expanduser().resolve()59    if not job_path.is_file():60        print(f"ERROR: job manifest does not exist: {job_path}", file=sys.stderr)61        return 262    job = json.loads(job_path.read_text(encoding="utf-8"))63    job_dir = job_path.parent64    errors: list[str] = []65    remote_blockers: list[str] = []66    warnings: list[str] = []67    media: dict[str, Any] = {}68 69    input_data = job.get("input", {})70    topic = str(input_data.get("topic", "")).strip()71    script_text = str(input_data.get("script_path", "")).strip()72    if not topic and not script_text:73        errors.append("topic or script_path is required")74    if script_text:75        require_file(script_text, "script", errors)76 77    image = require_file(str(input_data.get("presenter_image", "")), "presenter image", errors)78    if image:79        try:80            media["presenter_image"] = probe(image)81            streams = [82                stream83                for stream in media["presenter_image"].get("streams", [])84                if stream.get("codec_type") == "video"85            ]86            if not streams:87                errors.append("presenter image has no decodable image/video stream")88            else:89                width = int(streams[0].get("width") or 0)90                height = int(streams[0].get("height") or 0)91                if min(width, height) < 512:92                    warnings.append(f"presenter image is low resolution: {width}x{height}")93        except (subprocess.CalledProcessError, json.JSONDecodeError) as exc:94            errors.append(f"could not decode presenter image: {exc}")95 96    voice_value = str(input_data.get("voice_sample", "")).strip()97    if voice_value:98        voice = require_file(voice_value, "voice sample", errors)99        if voice:100            try:101                media["voice_sample"] = probe(voice)102                streams = [103                    stream104                    for stream in media["voice_sample"].get("streams", [])105                    if stream.get("codec_type") == "audio"106                ]107                if not streams:108                    errors.append("voice sample has no audio stream")109                duration = float(media["voice_sample"].get("format", {}).get("duration") or 0)110                if duration < 4:111                    warnings.append(f"voice sample is short: {duration:.3f}s")112                if duration > 60:113                    warnings.append(f"voice sample is unusually long: {duration:.3f}s")114            except (subprocess.CalledProcessError, json.JSONDecodeError, ValueError) as exc:115                errors.append(f"could not decode voice sample: {exc}")116        if not input_data.get("voice_clone_approved"):117            remote_blockers.append("voice_clone_approved must be true before voice cloning")118    else:119        warnings.append("no voice sample supplied; use and record a stock voice")120 121    supporting_reports = []122    for value in input_data.get("supporting_media", []):123        path = require_file(str(value), "supporting media", errors)124        if path:125            try:126                supporting_reports.append({"file": path.name, "probe": probe(path)})127            except (subprocess.CalledProcessError, json.JSONDecodeError) as exc:128                errors.append(f"could not decode supporting media {path}: {exc}")129    media["supporting_media"] = supporting_reports130 131    if not input_data.get("rights_confirmed"):132        remote_blockers.append("rights_confirmed must be true before presenter synthesis")133    if not input_data.get("adult_presenter_confirmed"):134        remote_blockers.append("adult_presenter_confirmed must be true before presenter synthesis")135    if not input_data.get("remote_upload_approved"):136        remote_blockers.append("remote_upload_approved must be true before remote generation")137 138    manual = job.get("manual_input_review", {})139    for key in ("image_viewed", "single_clear_face", "image_has_no_unwanted_text"):140        if not manual.get(key):141            errors.append(f"manual_input_review.{key} must be true")142    if voice_value:143        for key in ("voice_sample_listened", "single_clear_speaker"):144            if not manual.get(key):145                errors.append(f"manual_input_review.{key} must be true")146 147    creative = job.get("creative", {})148    duration = float(creative.get("duration_target_s") or 0)149    if not 5 <= duration <= 1800:150        errors.append("creative.duration_target_s must be between 5 and 1800")151    width = int(creative.get("width") or 0)152    height = int(creative.get("height") or 0)153    if min(width, height) < 256 or max(width, height) > 7680:154        errors.append("creative width/height must be between 256 and 7680")155    if int(creative.get("fps") or 0) not in (24, 25, 30, 50, 60):156        errors.append("creative.fps must be one of 24, 25, 30, 50, or 60")157 158    report_path = job_dir / "qa" / "reports" / "preflight.json"159    report_path.parent.mkdir(parents=True, exist_ok=True)160    report = {161        "ok": not errors,162        "remote_ready": not errors and not remote_blockers,163        "job": job_path.name,164        "errors": errors,165        "remote_blockers": remote_blockers,166        "warnings": warnings,167        "media": media,168    }169    write_atomic(report_path, json.dumps(report, ensure_ascii=False, indent=2) + "\n")170    job.setdefault("qa", {})["preflight_report"] = "qa/reports/preflight.json"171    write_atomic(job_path, json.dumps(job, ensure_ascii=False, indent=2) + "\n")172    print(json.dumps(report, ensure_ascii=False, indent=2))173    return 0 if not errors else 1174 175 176if __name__ == "__main__":177    raise SystemExit(main())178