# SPDX-License-Identifier: Apache-2.0 """Build a 9:16 comparison without trimming the generated video's timeline. Explicit half-open frame mappings determine the lower whitebox panel. Missing coverage is rejected. Declared unmatched spans remain visibly labelled; they never masquerade as matching motion. This builder adds no TTS or music. """ from __future__ import annotations import argparse from fractions import Fraction import hashlib import json import os from pathlib import Path import shutil import subprocess import tempfile def command_run(command, *, cwd=None, env=None, capture=False): options = dict(cwd=cwd, env=env, check=True, creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0) if capture: options.update(capture_output=True, text=True, encoding="utf-8") else: options.update(stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, encoding="utf-8") return subprocess.run(command, **options) def probe_media(path, ffprobe="ffprobe", env=None): result = command_run([ffprobe, "-v", "error", "-count_frames", "-show_entries", "stream=codec_type,codec_name,width,height,r_frame_rate,avg_frame_rate,nb_read_frames,duration", "-show_entries", "format=duration", "-of", "json", str(path)], env=env, capture=True) data = json.loads(result.stdout) videos = [stream for stream in data.get("streams", []) if stream.get("codec_type") == "video"] if len(videos) != 1: raise ValueError("Exactly one video stream is required") video = videos[0] if Fraction(video["avg_frame_rate"]) != Fraction(video["r_frame_rate"]): raise ValueError("Variable-frame-rate inputs are not supported; supply an explicitly normalized source") return {"frames": int(video["nb_read_frames"]), "fps": str(Fraction(video["avg_frame_rate"])), "width": video["width"], "height": video["height"], "video_duration_s": float(video["duration"]), "container_duration_s": float(data["format"]["duration"]), "audio_codecs": [s["codec_name"] for s in data["streams"] if s.get("codec_type") == "audio"], "streams": data["streams"]} def frame_range(value, maximum, name): if (not isinstance(value, list) or len(value) != 2 or any(type(number) is not int for number in value) or not 0 <= value[0] < value[1] <= maximum): raise ValueError(f"{name} must be an integer half-open range within [0, {maximum})") return value def validate_mapping(mapping, final_frames, whitebox_frames, fps): if mapping.get("range_semantics") != "half_open_frames": raise ValueError("Mapping must declare range_semantics=half_open_frames") if Fraction(str(mapping.get("fps", 0))) != Fraction(str(fps)): raise ValueError("Mapping fps must match both source videos") if mapping.get("final_video_frames", final_frames) != final_frames: raise ValueError("Declared final frame count disagrees with the actual full video") if mapping.get("whitebox_video_frames", whitebox_frames) != whitebox_frames: raise ValueError("Declared whitebox frame count disagrees with the actual source") segments = mapping.get("segments") if not isinstance(segments, list) or not segments: raise ValueError("Mapping must contain explicit segments") next_frame = 0 for number, segment in enumerate(segments): destination = frame_range(segment.get("destination"), final_frames, f"segment {number} destination") if destination[0] != next_frame: raise ValueError("Every final frame must be covered exactly once; gaps and overlaps are rejected") mode = segment.get("mode") if mode not in ("match", "unmatched_hold", "unmatched"): raise ValueError(f"Unsupported mapping mode: {mode}") if mode == "unmatched": if segment.get("source") is not None: raise ValueError("Unmatched blank spans must not claim a whitebox source") else: source = frame_range(segment.get("source"), whitebox_frames, f"segment {number} source") if mode == "unmatched_hold" and source[1] - source[0] != 1: raise ValueError("An unmatched hold must name exactly one reference frame") if mode == "match": ratio = (destination[1] - destination[0]) / (source[1] - source[0]) if not .9 <= ratio <= 1.1: raise ValueError("Match retiming exceeds 10%; declare unmatched coverage instead of stretching a whole clip") next_frame = destination[1] if next_frame != final_frames: raise ValueError("Mapping must preserve every final frame, including its last frame") return segments def reference_frame_at(segments, destination): for segment in segments: begin, end = segment["destination"] if begin <= destination < end: if segment["mode"] == "unmatched": return None first, last = segment["source"] if segment["mode"] == "unmatched_hold": return first return first + ((destination - begin) * (last - first)) // (end - begin) raise ValueError("Destination frame is outside the explicit mapping") def make_filter(width, height, panel_height, font_name, segments): top = round(height * .13) lower = round(height * .545) size = max(12, round(width * .038)) small = max(10, round(width * .029)) graph = [f"[0:v]setsar=1,scale={width}:{panel_height}:force_original_aspect_ratio=decrease:force_divisible_by=2," f"pad={width}:{panel_height}:(ow-iw)/2:(oh-ih)/2:color=0x12171d," f"pad={width}:{height}:0:{top}:color=0x12171d[top]", f"[top][1:v]overlay=0:{lower}:eof_action=pass[panels]"] def textfile(label, y, font_size): return f"drawtext=fontfile={font_name}:textfile={label}.txt:expansion=none:fontcolor=white:fontsize={font_size}:x=(w-tw)/2:y={y}" filters = [textfile("title", round(height * .035), size), textfile("final_label", round(height * .093), small), textfile("whitebox_label", round(height * .505), small), textfile("footer", round(height * .91), small)] for segment in segments: if segment["mode"] != "match": begin, end = segment["destination"] enabled = f"gte(n,{begin})*lt(n,{end})" filters.append(f"drawbox=x=0:y={lower + panel_height // 2 - size}:w=iw:h={size * 3}:color=black@0.85:t=fill:enable='{enabled}'") filters.append(textfile("unmatched", lower + panel_height // 2, small) + f":enable='{enabled}'") graph.append("[panels]" + ",".join(filters) + "[out]") return ";\n".join(graph) def resolve_executable(value): candidate = Path(value) found = str(candidate.resolve()) if candidate.is_file() else shutil.which(value) if not found: raise ValueError(f"Executable not found: {value}") return found def build(args): final = Path(args.final).resolve(strict=True) whitebox = Path(args.whitebox).resolve(strict=True) mapping_path = Path(args.mapping).resolve(strict=True) font = Path(args.font).resolve(strict=True) output = Path(args.out).resolve() if output.exists(): raise ValueError("Output exists; choose a new file to preserve existing results") if args.width < 180 or args.width % 18: raise ValueError("Canvas width must be at least 180 and a multiple of 18 for exact 9:16 output") ffmpeg = resolve_executable(args.ffmpeg) sibling = Path(ffmpeg).with_name("ffprobe.exe" if os.name == "nt" else "ffprobe") ffprobe = resolve_executable(args.ffprobe or (str(sibling) if sibling.is_file() else "ffprobe")) final_info, whitebox_info = probe_media(final, ffprobe), probe_media(whitebox, ffprobe) if final_info["fps"] != whitebox_info["fps"]: raise ValueError("Inputs must have the same constant fps; temporal normalization must be explicit") if any(codec not in ("aac", "mp3", "alac", "ac3", "eac3") for codec in final_info["audio_codecs"]): raise ValueError("Source audio is not supported for unchanged MP4 stream copy") mapping = json.loads(mapping_path.read_text(encoding="utf-8-sig")) segments = validate_mapping(mapping, final_info["frames"], whitebox_info["frames"], final_info["fps"]) if args.dry_run: return {"mode": "dry_run", "frames": final_info["frames"], "unmatched_spans": [s["destination"] for s in segments if s["mode"] != "match"]} output.parent.mkdir(parents=True, exist_ok=True) work_parent = Path(args.work_dir).resolve() if args.work_dir else output.parent work_parent.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix="comparison_", dir=work_parent) as raw: work = Path(raw) env = os.environ.copy() for key in ("TMP", "TEMP", "TMPDIR", "APPDATA", "LOCALAPPDATA", "XDG_CACHE_HOME"): env[key] = str(work) source_frames, mapped_frames = work / "reference", work / "mapped" source_frames.mkdir(); mapped_frames.mkdir() width, height = args.width, args.width * 16 // 9 panel_height = round(width * 9 / 16 / 2) * 2 fit = f"scale={width}:{panel_height}:force_original_aspect_ratio=decrease:force_divisible_by=2,pad={width}:{panel_height}:(ow-iw)/2:(oh-ih)/2:color=0x12171d,setsar=1" command_run([ffmpeg, "-hide_banner", "-v", "error", "-i", str(whitebox), "-vf", fit, "-fps_mode", "passthrough", "-start_number", "0", str(source_frames / "frame_%06d.png")], env=env) if len(list(source_frames.glob("frame_*.png"))) != whitebox_info["frames"]: raise RuntimeError("Decoded whitebox frame count changed") if any(s["mode"] == "unmatched" for s in segments): command_run([ffmpeg, "-hide_banner", "-v", "error", "-f", "lavfi", "-i", f"color=c=0x12171d:s={width}x{panel_height}:r=24", "-frames:v", "1", str(work / "blank.png")], env=env) for destination in range(final_info["frames"]): source_index = reference_frame_at(segments, destination) source = work / "blank.png" if source_index is None else source_frames / f"frame_{source_index:06d}.png" target = mapped_frames / f"frame_{destination:06d}.png" try: os.link(source, target) except OSError: shutil.copyfile(source, target) font_name = "font" + font.suffix.lower() shutil.copyfile(font, work / font_name) labels = {"title": "生成片与白模对照", "final_label": "生成片｜完整原速", "whitebox_label": "白模｜按实际镜头区间对应", "footer": "缺少参考的区间已明示，不作跟随成功证据", "unmatched": "生成延展｜本段无对应白模"} for name, label in labels.items(): (work / f"{name}.txt").write_text(label, encoding="utf-8") (work / "filter.txt").write_text(make_filter(width, height, panel_height, font_name, segments), encoding="utf-8") temporary_video = work / "comparison.mp4" command_run([ffmpeg, "-hide_banner", "-v", "error", "-i", str(final), "-framerate", final_info["fps"], "-start_number", "0", "-i", str(mapped_frames / "frame_%06d.png"), "-filter_complex_script", "filter.txt", "-map", "[out]", "-map", "0:a?", "-c:v", "libx264", "-crf", "18", "-preset", "medium", "-pix_fmt", "yuv420p", "-fps_mode", "passthrough", "-c:a", "copy", "-movflags", "+faststart", str(temporary_video)], cwd=work, env=env) actual = probe_media(temporary_video, ffprobe, env) if (actual["frames"] != final_info["frames"] or actual["fps"] != final_info["fps"] or [actual["width"], actual["height"]] != [width, height] or abs(actual["video_duration_s"] - final_info["video_duration_s"]) > .001 or abs(actual["container_duration_s"] - final_info["container_duration_s"]) > .002): raise RuntimeError("Comparison changed the full generated timeline or output geometry") command_run([ffmpeg, "-hide_banner", "-v", "error", "-xerror", "-i", str(temporary_video), "-f", "null", "-"], env=env) shutil.copyfile(temporary_video, output) report = {"schema_version": "whitebox-comparison-verification/1.0", "output": output.name, "sha256": hashlib.sha256(output.read_bytes()).hexdigest(), "source_final": final_info, "source_whitebox": whitebox_info, "actual_output": actual, "full_decode_pass": True, "final_timeline_preserved": True, "original_audio_stream_copy": bool(final_info["audio_codecs"]), "added_voice_or_music": False, "unmatched_spans": [s["destination"] for s in segments if s["mode"] != "match"], "mapping_sha256": hashlib.sha256(mapping_path.read_bytes()).hexdigest(), "builder_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "visual_review": "pending", "listening_review": "pending", "user_acceptance": "pending"} output.with_suffix(".verification.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return report def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--final", required=True) parser.add_argument("--whitebox", required=True) parser.add_argument("--mapping", required=True) parser.add_argument("--out", required=True) parser.add_argument("--font", required=True, help="Font file supporting the visible Chinese labels") parser.add_argument("--ffmpeg", default="ffmpeg") parser.add_argument("--ffprobe") parser.add_argument("--width", type=int, default=1080) parser.add_argument("--work-dir", help="Temporary files stay here; default is the output directory") parser.add_argument("--dry-run", action="store_true") args = parser.parse_args(argv) try: print(json.dumps(build(args), ensure_ascii=False, indent=2)) return 0 except (OSError, ValueError, RuntimeError, subprocess.CalledProcessError) as error: print(f"Comparison failed: {error}") if isinstance(error, subprocess.CalledProcessError) and error.stderr: print(error.stderr[-3000:]) return 1 if __name__ == "__main__": raise SystemExit(main())