#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json import subprocess from pathlib import Path ROOT = Path("/data/video-pipeline/HaTramAudio/project/020-Chiec-Nhan-Chi-Deo-Khi-Co-Nguoi-Nhin") PROJECT_ID = "020-Chiec-Nhan-Chi-Deo-Khi-Co-Nguoi-Nhin" RUN_ID = "run-20260721T124541Z-88551d8a" OWNER = "Levy" def sha(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for block in iter(lambda: f.read(8 * 1024 * 1024), b""): h.update(block) return h.hexdigest() def guard() -> None: lock = json.loads((ROOT / ".ownership-lock.json").read_text(encoding="utf-8")) for key, value in {"project_id": PROJECT_ID, "run_id": RUN_ID, "owner": OWNER}.items(): if lock.get(key) != value: raise RuntimeError(f"ownership mismatch: {key}") def read_verified(path: Path, allowed_statuses: tuple[str, ...] = ("completed",)) -> dict: data = json.loads(path.read_text(encoding="utf-8")) if data.get("project_id", PROJECT_ID) != PROJECT_ID or data.get("run_id") != RUN_ID: raise RuntimeError(f"receipt identity mismatch: {path}") if data.get("status") not in allowed_statuses or data.get("verified") is not True: raise RuntimeError(f"receipt not terminal/verified: {path}") return data def ensure_hash(path: Path, expected: str, label: str) -> None: if not path.is_file() or sha(path) != expected: raise RuntimeError(f"{label} artifact hash mismatch") def stream_signature(path: Path) -> tuple[dict, dict]: raw = subprocess.check_output([ "ffprobe", "-v", "error", "-show_entries", "stream=codec_name,profile,codec_type,width,height,pix_fmt,r_frame_rate,time_base,sample_rate,channels", "-of", "json", str(path), ]) streams = json.loads(raw).get("streams", []) videos = [item for item in streams if item.get("codec_type") == "video"] audios = [item for item in streams if item.get("codec_type") == "audio"] if len(videos) != 1 or len(audios) != 1: raise RuntimeError(f"expected one video and one audio stream: {path}") return videos[0], audios[0] def write_request(stage: str, payload: dict, bindings: dict) -> None: guard() work = ROOT / "work" / stage / RUN_ID work.mkdir(parents=True, exist_ok=True) request_path = work / "request.json" binding_path = work / "request-bindings.json" request_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") bindings["request_sha256"] = sha(request_path) binding_path.write_text(json.dumps(bindings, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"stage": stage, "request": str(request_path), "request_sha256": bindings["request_sha256"]}, ensure_ascii=False)) def footage() -> None: tts = read_verified(ROOT / "script" / "tts-manifest.json") layout = read_verified(ROOT / "image" / "layout-manifest.json") promotion = read_verified(ROOT / "script" / "promotion-report.json", ("passed", "completed", "completed_content_only")) canon_hash = promotion.get("canon_sha256") or promotion.get("source_canon_sha256") if tts.get("source_canon_sha256") != canon_hash or layout.get("source_canon_sha256") != canon_hash: raise RuntimeError("footage upstream canon mismatch") audio = ROOT / "audio" / "story-full.wav" layout_path = ROOT / "image" / "normalized" / "layout-1920x1080.png" ensure_hash(audio, tts["output_sha256"], "TTS") ensure_hash(layout_path, layout["layout_sha256"], "layout") payload = { "layout": str(layout_path), "footage_dir": "/data/video-pipeline/GacMaiAudio/footage", "pattern": "*.mp4", "recursive": True, "audio": str(audio), "output": str(ROOT / "output" / "footage" / "footage.mp4"), "seed": 2026072117, "overwrite": False, "width": 1920, "height": 1080, "fps": 30, "center_x": 656, "center_y": 0, "center_width": 608, "center_height": 1080, "video_encoder": "h264_nvenc", "preset": "p4", "cq": 21, } write_request("footage", payload, { "project_id": PROJECT_ID, "run_id": RUN_ID, "source_canon_sha256": canon_hash, "source_audio_sha256": tts["output_sha256"], "source_layout_sha256": layout["layout_sha256"], "mirror_required": True, "expected_horizontal_transform": "hflip", "source_audio_must_be_discarded": True, "production_narration_must_be_only_audio": True, "output": payload["output"], }) def final() -> None: intro = read_verified(ROOT / "log" / "intro-render.json") footage_receipt = read_verified(ROOT / "log" / "footage-render.json") manifest = json.loads((ROOT / "script" / "project-manifest.json").read_text(encoding="utf-8")) poster_event = manifest.get("gate_events", {}).get("poster", {}) layout_event = manifest.get("gate_events", {}).get("layout", {}) intro_path = ROOT / "output" / "intro" / "intro.mp4" footage_path = ROOT / "output" / "footage" / "footage.mp4" ensure_hash(intro_path, intro["artifact_sha256"], "intro") ensure_hash(footage_path, footage_receipt["artifact_sha256"], "footage") if not footage_receipt.get("technical_qa", {}).get("verified") or footage_receipt.get("content_qa") != "not_run_by_policy": raise RuntimeError("footage technical/no-content-QA contract is not verified") if intro["source_canon_sha256"] != footage_receipt["source_canon_sha256"]: raise RuntimeError("final upstream canon mismatch") if intro.get("source_poster_sha256") != poster_event.get("artifact_sha256"): raise RuntimeError("intro does not depend on current poster") if footage_receipt.get("source_layout_sha256") != layout_event.get("artifact_sha256"): raise RuntimeError("footage does not depend on current layout") intro_video, intro_audio = stream_signature(intro_path) footage_video, footage_audio = stream_signature(footage_path) video_fields = ("codec_name", "profile", "width", "height", "pix_fmt", "r_frame_rate", "time_base") audio_fields = ("codec_name", "profile", "sample_rate", "channels", "time_base") differences = { key: [intro_video.get(key), footage_video.get(key)] for key in video_fields if intro_video.get(key) != footage_video.get(key) } differences.update({ "audio_" + key: [intro_audio.get(key), footage_audio.get(key)] for key in audio_fields if intro_audio.get(key) != footage_audio.get(key) }) if differences: raise RuntimeError(f"stream-copy compatibility mismatch: {differences}") payload = { "intro_video_path": str(intro_path), "footage_video_path": str(footage_path), "output_video_path": str(ROOT / "output" / "final.mp4"), "packaging_mode": "stream_copy", "overwrite": False, "fps": 30, "width": 1920, "height": 1080, "video_encoder": "h264_nvenc", "preset": "p4", "cq": 21, } write_request("final", payload, { "project_id": PROJECT_ID, "run_id": RUN_ID, "source_canon_sha256": intro["source_canon_sha256"], "source_intro_sha256": intro["artifact_sha256"], "source_footage_sha256": footage_receipt["artifact_sha256"], "output": payload["output_video_path"], }) def transcode() -> None: final_receipt = read_verified(ROOT / "log" / "final-render.json") manifest = json.loads((ROOT / "script" / "project-manifest.json").read_text(encoding="utf-8")) poster_event = manifest.get("gate_events", {}).get("poster", {}) layout_event = manifest.get("gate_events", {}).get("layout", {}) final_path = ROOT / "output" / "final.mp4" ensure_hash(final_path, final_receipt["artifact_sha256"], "final") if not final_receipt.get("technical_qa", {}).get("verified") or final_receipt.get("content_qa") != "not_run_by_policy": raise RuntimeError("final technical/no-content-QA contract is not verified") if final_receipt.get("source_poster_sha256") != poster_event.get("artifact_sha256"): raise RuntimeError("final does not depend on current poster") if final_receipt.get("source_layout_sha256") != layout_event.get("artifact_sha256"): raise RuntimeError("final does not depend on current layout") payload = { "input_video_path": str(final_path), "output_video_path": str(ROOT / "output" / "final-upload.mp4"), "target_output_bytes": 800000000, "max_output_bytes": 1000000000, "width": 1920, "height": 1080, "fps": 30, "audio_bitrate_kbps": 96, "overwrite": False, } write_request("transcode", payload, { "project_id": PROJECT_ID, "run_id": RUN_ID, "source_canon_sha256": final_receipt["source_canon_sha256"], "source_final_sha256": final_receipt["artifact_sha256"], "output": payload["output_video_path"], }) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("stage", choices=["footage", "final", "transcode"]) args = parser.parse_args() guard() {"footage": footage, "final": final, "transcode": transcode}[args.stage]() if __name__ == "__main__": main()