skills/fountain-clip-producer/modules/qa/scripts/validate-clip.py

#!/usr/bin/env python3
"""The blocking QA gate for a produced clip. Aggregates whole-clip checks
(dimensions, fps, duration, audio presence, black-frame regression) with the
per-module reports handed up from framing (visual QA) and captions (fit
report) into a single qa_report.json. Does not redo those per-module checks —
only confirms they exist and passed.
"""

import argparse
import json
import math
import re
import subprocess
import sys
from pathlib import Path


def run(cmd):
    return subprocess.run(cmd, check=False, text=True, capture_output=True)


def ffprobe(path, count_frames=False):
    cmd = ["ffprobe", "-hide_banner", "-v", "error"]
    if count_frames:
        cmd.append("-count_frames")
    cmd += [
        "-show_entries",
        "format=duration,size:stream=index,codec_type,codec_name,width,height,r_frame_rate,duration,nb_read_frames,pix_fmt",
        "-of",
        "json",
        str(path),
    ]
    proc = run(cmd)
    if proc.returncode != 0:
        raise RuntimeError(proc.stderr.strip() or f"ffprobe failed for {path}")
    return json.loads(proc.stdout)


def parse_fps(value):
    if not value or value == "0/0":
        return None
    if "/" in value:
        num, den = value.split("/", 1)
        den_value = float(den)
        return float(num) / den_value if den_value else None
    return float(value)


def video_stream(probe):
    for stream in probe.get("streams", []):
        if stream.get("codec_type") == "video":
            return stream
    return None


def audio_streams(probe):
    return [stream for stream in probe.get("streams", []) if stream.get("codec_type") == "audio"]


def blackdetect(path):
    proc = run(
        ["ffmpeg", "-hide_banner", "-i", str(path), "-vf", "blackdetect=d=0.05:pix_th=0.10", "-an", "-f", "null", "-"]
    )
    text = f"{proc.stdout}\n{proc.stderr}"
    intervals = []
    pattern = re.compile(r"black_start:([0-9.]+) black_end:([0-9.]+) black_duration:([0-9.]+)")
    for match in pattern.finditer(text):
        intervals.append(
            {"start": float(match.group(1)), "end": float(match.group(2)), "duration": float(match.group(3))}
        )
    return intervals


def covered(interval, baselines, tolerance=0.04):
    return any(
        interval["start"] >= baseline["start"] - tolerance and interval["end"] <= baseline["end"] + tolerance
        for baseline in baselines
    )


def add_check(checks, name, passed, detail):
    checks.append({"name": name, "status": "pass" if passed else "fail", "detail": detail})


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--clean-master", required=True)
    parser.add_argument("--final", required=True)
    parser.add_argument("--caption-layer")
    parser.add_argument("--caption-fit-report")
    parser.add_argument("--expected-width", type=int, required=True)
    parser.add_argument("--expected-height", type=int, required=True)
    parser.add_argument("--expected-fps", type=float, required=True)
    parser.add_argument("--expected-duration", type=float, required=True)
    parser.add_argument(
        "--landscape-master",
        help="clip-landscape-master.mp4 from module media. Its height is the real detail behind the export.",
    )
    parser.add_argument(
        "--max-upscale",
        type=float,
        default=2.0,
        help="Largest allowed ratio of delivered height to clean-master height. 2.0 passes a 1080p source "
        "for a 1920-tall delivery and fails a 720p one.",
    )
    parser.add_argument("--contact-sheet")
    parser.add_argument("--visual-report", help="visual_qa_report.json produced by the framing module")
    parser.add_argument("--report", required=True)
    args = parser.parse_args()

    checks = []
    report = {
        "status": "fail",
        "cleanMaster": args.clean_master,
        "final": args.final,
        "captionLayer": args.caption_layer,
        "captionFitReport": args.caption_fit_report,
        "checks": checks,
    }

    try:
        clean_probe = ffprobe(args.clean_master, count_frames=True)
        final_probe = ffprobe(args.final, count_frames=True)
        clean_video = video_stream(clean_probe)
        final_video = video_stream(final_probe)
        if not clean_video or not final_video:
            raise RuntimeError("missing video stream")

        final_fps = parse_fps(final_video.get("r_frame_rate"))
        final_duration = float(final_probe.get("format", {}).get("duration", 0))
        final_frames = int(final_video.get("nb_read_frames", 0))
        expected_frames = int(round(args.expected_duration * args.expected_fps))

        add_check(
            checks,
            "final_dimensions",
            final_video.get("width") == args.expected_width and final_video.get("height") == args.expected_height,
            {
                "actual": [final_video.get("width"), final_video.get("height")],
                "expected": [args.expected_width, args.expected_height],
            },
        )
        # A vertical crop keeps the whole height of the landscape master, so
        # that height is the real detail behind the delivered one. The container
        # reports 1080x1920 either way, which is how a 720p source shipped as
        # 1080p until this check existed. The clean master is already vertical,
        # so it cannot answer this -- only the landscape one can.
        if args.landscape_master:
            landscape_video = video_stream(ffprobe(args.landscape_master)) or {}
            source_height = landscape_video.get("height") or 0
            upscale = (args.expected_height / source_height) if source_height else None
            add_check(
                checks,
                "source_resolution",
                upscale is not None and upscale <= args.max_upscale,
                {
                    "sourceHeight": source_height,
                    "finalHeight": args.expected_height,
                    "upscale": round(upscale, 2) if upscale else None,
                    "maxUpscale": args.max_upscale,
                },
            )
        add_check(
            checks,
            "final_fps",
            final_fps is not None and math.isclose(final_fps, args.expected_fps, abs_tol=0.01),
            {"actual": final_video.get("r_frame_rate"), "expected": args.expected_fps},
        )
        add_check(
            checks,
            "final_duration",
            math.isclose(final_duration, args.expected_duration, abs_tol=0.25),
            {"actual": final_duration, "expected": args.expected_duration},
        )
        add_check(
            checks,
            "final_frame_count",
            abs(final_frames - expected_frames) <= 1,
            {"actual": final_frames, "expected": expected_frames},
        )
        add_check(
            checks,
            "final_audio_present",
            bool(audio_streams(final_probe)),
            {"audioStreams": len(audio_streams(final_probe))},
        )

        if args.caption_layer:
            layer_probe = ffprobe(args.caption_layer, count_frames=True)
            layer_video = video_stream(layer_probe)
            if not layer_video:
                raise RuntimeError("caption layer missing video stream")
            layer_fps = parse_fps(layer_video.get("r_frame_rate"))
            layer_duration = float(layer_video.get("duration") or layer_probe.get("format", {}).get("duration", 0))
            layer_frames = int(layer_video.get("nb_read_frames", 0))
            add_check(
                checks,
                "caption_layer_dimensions",
                layer_video.get("width") == args.expected_width and layer_video.get("height") == args.expected_height,
                {
                    "actual": [layer_video.get("width"), layer_video.get("height")],
                    "expected": [args.expected_width, args.expected_height],
                },
            )
            add_check(
                checks,
                "caption_layer_alpha",
                layer_video.get("pix_fmt") in {"argb", "rgba", "yuva420p", "yuva444p"},
                {"actual": layer_video.get("pix_fmt")},
            )
            add_check(
                checks,
                "caption_layer_fps",
                layer_fps is not None and math.isclose(layer_fps, args.expected_fps, abs_tol=0.01),
                {"actual": layer_video.get("r_frame_rate"), "expected": args.expected_fps},
            )
            add_check(
                checks,
                "caption_layer_duration",
                math.isclose(layer_duration, args.expected_duration, abs_tol=0.05),
                {"actual": layer_duration, "expected": args.expected_duration},
            )
            add_check(
                checks,
                "caption_layer_frame_count",
                abs(layer_frames - expected_frames) <= 1,
                {"actual": layer_frames, "expected": expected_frames},
            )

        if args.caption_fit_report:
            fit_path = Path(args.caption_fit_report)
            fit_report = json.loads(fit_path.read_text()) if fit_path.exists() else []
            violations = []
            for item in fit_report:
                max_width = int(item.get("maxWidth", 0))
                line_widths = item.get("lineWidths", [])
                widest = max(line_widths) if line_widths else 0
                if widest > max_width:
                    violations.append(
                        {
                            "text": item.get("text"),
                            "plate": item.get("plate"),
                            "maxWidth": max_width,
                            "widestLine": widest,
                            "lineWidths": line_widths,
                        }
                    )
            report["captionFit"] = {"path": str(fit_path), "items": len(fit_report), "violations": violations}
            point_sizes = sorted({item.get("pointSize") for item in fit_report if item.get("pointSize")})
            add_check(
                checks,
                "caption_text_fit",
                fit_path.exists() and bool(fit_report) and not violations,
                {"path": str(fit_path), "items": len(fit_report), "violations": violations},
            )
            add_check(
                checks,
                "caption_font_size_consistency",
                len(point_sizes) == 1,
                {"path": str(fit_path), "pointSizes": point_sizes},
            )

        clean_black = blackdetect(args.clean_master)
        final_black = blackdetect(args.final)
        introduced_black = [interval for interval in final_black if not covered(interval, clean_black)]
        report["blackdetect"] = {"clean": clean_black, "final": final_black, "introduced": introduced_black}
        add_check(checks, "no_new_black_intervals", not introduced_black, {"introduced": introduced_black})

        if args.contact_sheet:
            contact_sheet = Path(args.contact_sheet)
            add_check(
                checks,
                "contact_sheet_exists",
                contact_sheet.exists() and contact_sheet.stat().st_size > 0,
                {"path": str(contact_sheet)},
            )

        if args.visual_report:
            visual_path = Path(args.visual_report)
            visual = json.loads(visual_path.read_text()) if visual_path.exists() else {"status": "missing"}
            report["visualPersonQa"] = visual
            add_check(
                checks,
                "visual_person_qa",
                visual.get("status") == "pass",
                {
                    "path": str(visual_path),
                    "status": visual.get("status"),
                    "missingCount": visual.get("missingCount"),
                    "missing": visual.get("missing", []),
                },
            )

        report["status"] = "pass" if all(check["status"] == "pass" for check in checks) else "fail"
    except Exception as exc:
        report["error"] = str(exc)

    Path(args.report).write_text(json.dumps(report, indent=2) + "\n")
    print(json.dumps(report, indent=2))
    return 0 if report["status"] == "pass" else 1


if __name__ == "__main__":
    sys.exit(main())