#!/usr/bin/env python3 from __future__ import annotations import hashlib import json import os import subprocess import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] PROJECT_MANIFEST = ROOT / "script/project-manifest.json" def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def resolve_active(manifest: dict, key: str) -> Path: value = (manifest.get("active_paths") or {}).get(key) if not isinstance(value, str) or not value: raise RuntimeError(f"active_paths.{key} is missing") path = Path(value) if not path.is_absolute(): path = ROOT / path try: path.resolve().relative_to(ROOT.resolve()) except ValueError as exc: raise RuntimeError(f"active path escapes project: {key}") from exc return path def atomic_json(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) data = (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8") fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: with os.fdopen(fd, "wb") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) os.replace(name, path) finally: if os.path.exists(name): os.unlink(name) def run(command: list[str]) -> None: subprocess.run(command, check=True) def main() -> int: project = json.loads(PROJECT_MANIFEST.read_text(encoding="utf-8")) paths = { key: resolve_active(project, key) for key in ( "left_panel_master", "right_panel_master", "intro_master", "left_panel_normalized", "right_panel_normalized", "intro_normalized", "layout", "artwork_visual_receipt", "orientation_receipt", "promotion_receipt", "layout_normalization_receipt", ) } required = [ paths["left_panel_master"], paths["right_panel_master"], paths["intro_master"], paths["artwork_visual_receipt"], paths["orientation_receipt"], paths["promotion_receipt"], ] missing = [str(path) for path in required if not path.exists()] if missing: raise RuntimeError(f"normalization inputs missing: {missing}") promotion = json.loads(paths["promotion_receipt"].read_text(encoding="utf-8")) visual = json.loads(paths["artwork_visual_receipt"].read_text(encoding="utf-8")) orientation = json.loads(paths["orientation_receipt"].read_text(encoding="utf-8")) canon_hash = promotion.get("canon_sha256") if promotion.get("status") != "completed" or promotion.get("verified") is not True: raise RuntimeError("promotion is not verified") masters = { "left_panel": paths["left_panel_master"], "right_panel": paths["right_panel_master"], "intro": paths["intro_master"], } master_hashes = {name: sha256(path) for name, path in masters.items()} for gate_name, gate in (("visual", visual), ("orientation", orientation)): if gate.get("status") != "completed" or gate.get("verified") is not True: raise RuntimeError(f"{gate_name} gate is absent or failed") if gate.get("source_canon_sha256") != canon_hash: raise RuntimeError(f"{gate_name} gate is stale for canon") if gate.get("artifact_sha256") != master_hashes: raise RuntimeError(f"artwork master drift after {gate_name} gate") outputs = [ paths["left_panel_normalized"], paths["right_panel_normalized"], paths["intro_normalized"], paths["layout"], paths["layout_normalization_receipt"], ] existing = [str(path) for path in outputs if path.exists()] if existing: raise RuntimeError(f"normalized outputs already exist and require audit/invalidation: {existing}") for path in outputs: path.parent.mkdir(parents=True, exist_ok=True) # Scale down and pad only; never crop or redraw provider-rendered typography. run([ "ffmpeg", "-v", "error", "-y", "-i", str(paths["left_panel_master"]), "-vf", "scale=656:1080:force_original_aspect_ratio=decrease,pad=656:1080:(ow-iw)/2:(oh-ih)/2:color=#101820", "-frames:v", "1", str(paths["left_panel_normalized"]), ]) run([ "ffmpeg", "-v", "error", "-y", "-i", str(paths["right_panel_master"]), "-vf", "scale=656:1080:force_original_aspect_ratio=decrease,pad=656:1080:(ow-iw)/2:(oh-ih)/2:color=#101820", "-frames:v", "1", str(paths["right_panel_normalized"]), ]) run([ "ffmpeg", "-v", "error", "-y", "-i", str(paths["intro_master"]), "-vf", "scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2:color=#101820", "-frames:v", "1", str(paths["intro_normalized"]), ]) run([ "ffmpeg", "-v", "error", "-y", "-i", str(paths["left_panel_normalized"]), "-i", str(paths["right_panel_normalized"]), "-filter_complex", "color=c=#101820:s=1920x1080:r=1[base];[base][0:v]overlay=0:0[tmp];[tmp][1:v]overlay=1264:0,format=rgb24[out]", "-map", "[out]", "-frames:v", "1", str(paths["layout"]), ]) result = { "status": "completed_pending_target_visual_review", "verified": False, "source_canon_sha256": canon_hash, "master_artifact_sha256": master_hashes, "visual_receipt": {"path": str(paths["artwork_visual_receipt"]), "sha256": sha256(paths["artwork_visual_receipt"])}, "orientation_receipt": {"path": str(paths["orientation_receipt"]), "sha256": sha256(paths["orientation_receipt"])}, "outputs": { str(path.relative_to(ROOT)): {"sha256": sha256(path), "bytes": path.stat().st_size} for path in outputs[:-1] }, "method": "mechanical scale-decrease and pad; no crop, text overlay, or text editing", "target_visual_review_required": True, } atomic_json(paths["layout_normalization_receipt"], result) print(json.dumps({"status": result["status"], "verified": False, "receipt": str(paths["layout_normalization_receipt"])}, ensure_ascii=False)) return 0 if __name__ == "__main__": try: sys.exit(main()) except Exception as exc: print(json.dumps({"status": "failed", "verified": False, "error": f"{type(exc).__name__}: {exc}"}, ensure_ascii=False), file=sys.stderr) sys.exit(1)