#!/usr/bin/env python3 import argparse import datetime import hashlib import json import os import subprocess from pathlib import Path ROOT = Path(__file__).resolve().parents[1] MANIFEST = ROOT / "script/project-manifest.json" def sha256(path): digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(8 * 1024 * 1024), b""): digest.update(block) return digest.hexdigest() def resolve(manifest, key): value = manifest.get("active_paths", {}).get(key) if not value: raise RuntimeError(f"active_paths.{key} missing") path = Path(value) path = path if path.is_absolute() else ROOT / path path = path.resolve() path.relative_to(ROOT.resolve()) return path def load_receipt(manifest, key, canon): path = resolve(manifest, key) value = json.loads(path.read_text(encoding="utf-8")) if value.get("status") != "completed" or value.get("verified") is not True: raise RuntimeError(f"{key} not terminal verified") receipt_canon = value.get("source_canon_sha256") if receipt_canon != canon: raise RuntimeError(f"{key} canon mismatch") return path, value def probe(path): result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration,size,format_name", "-show_entries", "stream=codec_type,codec_name,width,height,sample_rate,channels", "-of", "json", str(path)], capture_output=True, text=True, check=True, ) return json.loads(result.stdout) def atomic_json(path, value): part = path.with_suffix(path.suffix + ".part") part.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") os.replace(part, path) def final_qa(project): canon = project["canon_sha256"] final_receipt_path, final_receipt = load_receipt(project, "final_receipt", canon) intro_receipt_path, intro_receipt = load_receipt(project, "intro_receipt", canon) footage_receipt_path, footage_receipt = load_receipt(project, "footage_receipt", canon) artwork_path, artwork = load_receipt(project, "artwork_visual_receipt", canon) output = resolve(project, "final_master") poster = resolve(project, "intro_poster") if sha256(output) != final_receipt.get("artifact_sha256"): raise RuntimeError("final artifact hash drift") normalized_intro = (artwork.get("normalized_gates") or {}).get("intro") or {} if normalized_intro.get("verified") is not True or normalized_intro.get("sha256") != sha256(poster): raise RuntimeError("intro poster visual lineage failed") if intro_receipt.get("poster_sha256") != sha256(poster): raise RuntimeError("intro video does not reference current poster") info = probe(output) streams = info.get("streams", []) videos = [row for row in streams if row.get("codec_type") == "video"] audios = [row for row in streams if row.get("codec_type") == "audio"] duration = float(info["format"]["duration"]) expected = float(intro_receipt["duration_seconds"]) + float(footage_receipt["duration_seconds"]) checks = { "final_hash": True, "one_h264_video": len(videos) == 1 and videos[0].get("codec_name") == "h264", "one_aac_audio": len(audios) == 1 and audios[0].get("codec_name") == "aac", "resolution_1920x1080": len(videos) == 1 and (videos[0].get("width"), videos[0].get("height")) == (1920, 1080), "duration_matches_intro_plus_footage": abs(duration - expected) <= 0.08, "endpoint_verified": final_receipt.get("endpoint_verification", {}).get("verified") is True, "intro_visual_provider_still_verified": normalized_intro.get("verified") is True, "intro_poster_hash_linked": intro_receipt.get("poster_sha256") == normalized_intro.get("sha256"), "footage_integrity_verified": footage_receipt.get("mirror_applied") is True and footage_receipt.get("horizontal_mirror") == "hflip" and footage_receipt.get("source_audio_discarded") is True, "no_footage_visual_evidence": True, } if not all(checks.values()): raise RuntimeError(f"final QA failed: {checks}") now = datetime.datetime.now(datetime.timezone.utc).isoformat() receipt_path = resolve(project, "final_qa_receipt") receipt = { "status": "completed", "verified": True, "source_canon_sha256": canon, "scope": "machine integrity for final plus direct visual verification of provider-rendered intro poster", "footage_visual_review_performed": False, "video_frame_extraction_performed": False, "intro_visual_evidence": {"path": str(poster.relative_to(ROOT)), "sha256": sha256(poster), "artwork_visual_receipt": str(artwork_path.relative_to(ROOT)), "artwork_visual_receipt_sha256": sha256(artwork_path)}, "inputs": { "final_receipt": {"path": str(final_receipt_path.relative_to(ROOT)), "sha256": sha256(final_receipt_path)}, "intro_receipt": {"path": str(intro_receipt_path.relative_to(ROOT)), "sha256": sha256(intro_receipt_path)}, "footage_receipt": {"path": str(footage_receipt_path.relative_to(ROOT)), "sha256": sha256(footage_receipt_path)}, }, "output_path": str(output.relative_to(ROOT)), "artifact_sha256": sha256(output), "duration_seconds": duration, "duration_basis": {"intro": intro_receipt["duration_seconds"], "footage": footage_receipt["duration_seconds"], "expected_sum": expected}, "independent_probe": info, "checks": checks, "completed_at": now, } if receipt_path.exists(): raise RuntimeError("final QA receipt already exists") atomic_json(receipt_path, receipt) latest = json.loads(MANIFEST.read_text(encoding="utf-8")) latest["final_qa"] = {"status": "completed", "verified": True, "receipt": str(receipt_path.relative_to(ROOT)), "artifact_sha256": receipt["artifact_sha256"], "completed_at": now} latest["updated_at"] = now atomic_json(MANIFEST, latest) print(json.dumps({"status": "completed", "verified": True, "stage": "final_qa", "receipt": str(receipt_path), "artifact_sha256": receipt["artifact_sha256"]}, ensure_ascii=False)) def publish_ready(project): canon = project["canon_sha256"] receipt_keys = [ "promotion_receipt", "tts_receipt", "artwork_generation_receipt", "artwork_visual_receipt", "layout_receipt", "intro_receipt", "footage_receipt", "final_receipt", "final_qa_receipt", "transcode_receipt", "metadata_receipt", ] records = {} for key in receipt_keys: path, value = load_receipt(project, key, canon) records[key] = {"path": str(path.relative_to(ROOT)), "sha256": sha256(path)} upload = resolve(project, "final_upload") thumbnail = resolve(project, "intro_poster") metadata = resolve(project, "metadata") transcode = json.loads(resolve(project, "transcode_receipt").read_text(encoding="utf-8")) metadata_receipt = json.loads(resolve(project, "metadata_receipt").read_text(encoding="utf-8")) final_qa_receipt = json.loads(resolve(project, "final_qa_receipt").read_text(encoding="utf-8")) if not upload.is_file() or upload.stat().st_size <= 0 or upload.stat().st_size >= 1_000_000_000: raise RuntimeError("upload copy absent or not under one billion bytes") if sha256(upload) != transcode.get("artifact_sha256") or upload.stat().st_size != transcode.get("artifact_bytes"): raise RuntimeError("upload copy transcode lineage failed") if transcode.get("endpoint_verification", {}).get("under_one_gb") is not True: raise RuntimeError("endpoint under-one-GB verification missing") if sha256(metadata) != metadata_receipt.get("artifact_sha256"): raise RuntimeError("metadata hash drift") if final_qa_receipt.get("video_frame_extraction_performed") is not False or final_qa_receipt.get("footage_visual_review_performed") is not False: raise RuntimeError("forbidden footage visual evidence detected") artifacts = { "upload": {"path": str(upload.relative_to(ROOT)), "sha256": sha256(upload), "bytes": upload.stat().st_size}, "thumbnail": {"path": str(thumbnail.relative_to(ROOT)), "sha256": sha256(thumbnail), "bytes": thumbnail.stat().st_size}, "metadata": {"path": str(metadata.relative_to(ROOT)), "sha256": sha256(metadata), "bytes": metadata.stat().st_size}, } now = datetime.datetime.now(datetime.timezone.utc).isoformat() output = resolve(project, "publish_ready_receipt") if output.exists(): raise RuntimeError("publish-ready receipt already exists") receipt = { "status": "completed", "verified": True, "source_canon_sha256": canon, "active_version": project.get("active_version"), "receipts": records, "artifacts": artifacts, "checks": { "all_receipts_terminal_same_canon": True, "upload_copy_under_one_billion_bytes": True, "upload_copy_hash_linked": True, "metadata_hash_linked": True, "thumbnail_hash_locked": True, "footage_visual_qa_absent": True, "publish_block_absent": not any(ROOT.glob("DO_NOT_PUBLISH*")) and not (ROOT / "publish_block").exists(), }, "completed_at": now, } if not all(receipt["checks"].values()): raise RuntimeError(f"publish-ready blockers: {receipt['checks']}") atomic_json(output, receipt) latest = json.loads(MANIFEST.read_text(encoding="utf-8")) latest["steps"]["publish_ready"] = "completed" latest["publish_ready"] = {"status": "completed", "verified": True, "receipt": str(output.relative_to(ROOT)), "upload_sha256": artifacts["upload"]["sha256"], "completed_at": now} latest["updated_at"] = now atomic_json(MANIFEST, latest) print(json.dumps({"status": "completed", "verified": True, "stage": "publish_ready", "receipt": str(output), "upload": artifacts["upload"]}, ensure_ascii=False)) def main(): parser = argparse.ArgumentParser() parser.add_argument("stage", choices=("final-qa", "publish-ready")) args = parser.parse_args() project = json.loads(MANIFEST.read_text(encoding="utf-8")) final_qa(project) if args.stage == "final-qa" else publish_ready(project) if __name__ == "__main__": main()