#!/usr/bin/env python3 from __future__ import annotations import hashlib import json import os import subprocess from datetime import datetime, timezone from pathlib import Path ROOT = Path("/data/video-pipeline/HaTramAudio/project/029-Chiec-O-De-Lai-O-Tram-Xe-Buyt-Cuoi") PROJECT_ID = "029-Chiec-O-De-Lai-O-Tram-Xe-Buyt-Cuoi" RUN_ID = "run-20260722T162951Z-87a939ad" OWNER = "Levy" PROVIDER = ROOT / "image" / "provider" NORMALIZED = ROOT / "image" / "normalized" ARTWORK_QA = ROOT / "log" / "artwork-qa.json" PROMPTS = ROOT / "script" / "image-prompts.json" PROMOTION = ROOT / "script" / "promotion-report.json" PROJECT_MANIFEST = ROOT / "script" / "project-manifest.json" MANIFEST = ROOT / "image" / "layout-manifest.json" TARGETS = { "left_panel": (PROVIDER / "left-panel-provider.png", NORMALIZED / "left-panel-656x1080.png", 656, 1080), "right_panel": (PROVIDER / "right-panel-provider.png", NORMALIZED / "right-panel-656x1080.png", 656, 1080), "intro_poster": (PROVIDER / "intro-poster-provider.png", NORMALIZED / "intro-poster-1920x1080.png", 1920, 1080), } def now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") 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: d = 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 d.get(key) != value: raise RuntimeError(f"ownership mismatch: {key}") def dims(path: Path) -> tuple[int, int]: raw = subprocess.check_output([ "ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height", "-of", "json", str(path), ]) streams = json.loads(raw).get("streams", []) if len(streams) != 1: raise RuntimeError(f"invalid image stream: {path}") return int(streams[0]["width"]), int(streams[0]["height"]) def atomic_json(path: Path, value: dict) -> None: guard() tmp = path.with_name("." + path.name + ".tmp") tmp.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") os.replace(tmp, path) def main() -> None: guard() project_manifest = json.loads(PROJECT_MANIFEST.read_text(encoding="utf-8")) story = project_manifest.get("gate_events", {}).get("story", {}) if project_manifest.get("steps", {}).get("story") != "completed" or story.get("verified") is not True: raise RuntimeError("Story Gate is not committed") qa = json.loads(ARTWORK_QA.read_text(encoding="utf-8")) prompts = json.loads(PROMPTS.read_text(encoding="utf-8")) promotion = json.loads(PROMOTION.read_text(encoding="utf-8")) canon_hash = promotion.get("canon_sha256") or promotion.get("source_canon_sha256") if project_manifest.get("canon_sha256") != canon_hash or story.get("canon_sha256") != canon_hash: raise RuntimeError("Story Gate canon binding mismatch") if not qa.get("verified") or qa.get("status") != "passed": raise RuntimeError("artwork visual QA is not passed") if qa.get("source_canon_sha256") != canon_hash or prompts.get("source_canon_sha256") != canon_hash: raise RuntimeError("artwork authority canon mismatch") NORMALIZED.mkdir(parents=True, exist_ok=True) items = {} for key, (source, target, tw, th) in TARGETS.items(): if not source.is_file(): raise RuntimeError(f"missing provider artifact: {source}") expected = qa.get("items", {}).get(key, {}).get("artifact_sha256") source_hash = sha(source) if source_hash != expected: raise RuntimeError(f"provider hash mismatch: {key}") sw, sh = dims(source) source_ratio = sw / sh target_ratio = tw / th ratio_delta = abs(source_ratio - target_ratio) / target_ratio if ratio_delta > 0.065: raise RuntimeError(f"aspect ratio too far for full-frame resample: {key} delta={ratio_delta:.6f}") tmp = target.with_name("." + target.name + ".tmp.png") subprocess.check_call([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(source), "-vf", f"scale={tw}:{th}:flags=lanczos", "-frames:v", "1", str(tmp), ]) if dims(tmp) != (tw, th): raise RuntimeError(f"normalized dimensions mismatch: {key}") guard() os.replace(tmp, target) items[key] = { "source_path": str(source), "source_sha256": source_hash, "source_dimensions": [sw, sh], "target_path": str(target), "target_sha256": sha(target), "target_dimensions": [tw, th], "ratio_delta": ratio_delta, "operation": "full_frame_resample_only", "crop": False, "pad": False, "rotate": False, "typography_postprocess": False, } layout = NORMALIZED / "layout-1920x1080.png" tmp_layout = layout.with_name("." + layout.name + ".tmp.png") subprocess.check_call([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(TARGETS["left_panel"][1]), "-i", str(TARGETS["right_panel"][1]), "-filter_complex", "[0:v]format=rgb24[left];[1:v]format=rgb24[right];color=c=#101820:s=1920x1080:r=1[base];[base][left]overlay=0:0[tmp];[tmp][right]overlay=1264:0,format=rgb24[out]", "-map", "[out]", "-frames:v", "1", str(tmp_layout), ]) if dims(tmp_layout) != (1920, 1080): raise RuntimeError("layout dimensions mismatch") guard() os.replace(tmp_layout, layout) manifest = { "schema_version": 1, "project_id": PROJECT_ID, "run_id": RUN_ID, "status": "qa_pending", "verified": False, "technical_verified": True, "visual_qa": "pending", "builder_location": "agent_local", "builder_tool": "ffmpeg", "normalized_config": { "resample": "scale=::flags=lanczos", "canvas": "1920x1080 rgb24", "layout_filter": "left@0,0; center=#101820@656,0,608x1080; right@1264,0", "text_overlay": False, }, "source_canon_sha256": canon_hash, "artwork_qa_sha256": sha(ARTWORK_QA), "provider_only_typography": True, "items": items, "layout_path": str(layout), "layout_sha256": sha(layout), "layout_dimensions": [1920, 1080], "coordinates": {"left": [0, 0, 656, 1080], "center": [656, 0, 608, 1080], "right": [1264, 0, 656, 1080]}, "center_fill": "#101820", "created_at": now(), } atomic_json(MANIFEST, manifest) print(json.dumps({"status": "completed", "verified": True, "layout": str(layout), "layout_sha256": manifest["layout_sha256"]}, ensure_ascii=False)) if __name__ == "__main__": main()