#!/usr/bin/env python3 from __future__ import annotations import datetime import hashlib import json import os import subprocess import sys import tempfile import time import urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parents[1] MANIFEST = ROOT / "script/project-manifest.json" BASE = "http://192.168.1.104:8023" 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(manifest: dict, key: str) -> Path: value = (manifest.get("active_paths") or {}).get(key) if not isinstance(value, str) or not value: raise RuntimeError(f"missing active_paths.{key}") path = Path(value) if not path.is_absolute(): path = ROOT / path path.resolve().relative_to(ROOT.resolve()) 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(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp") 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 request_json(url: str, payload: dict | None = None) -> dict: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else None request = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"} if data else {}, method="POST" if data else "GET") with urllib.request.urlopen(request, timeout=60) as response: return json.loads(response.read().decode("utf-8")) def probe(path: Path) -> dict: return json.loads(subprocess.check_output([ "ffprobe", "-v", "error", "-show_entries", "format=duration,size:stream=codec_name,codec_type,width,height,sample_rate,channels", "-of", "json", str(path), ], text=True)) def duration(probe_value: dict) -> float: return float((probe_value.get("format") or {}).get("duration", 0)) def main() -> int: project = json.loads(MANIFEST.read_text(encoding="utf-8")) keys = ("intro", "intro_receipt", "intro_qa_receipt", "footage", "footage_receipt", "footage_qa_receipt", "final_master", "final_receipt") paths = {key: resolve(project, key) for key in keys} canon = project.get("canon_sha256") inputs = {} for stage in ("intro", "footage"): render = json.loads(paths[f"{stage}_receipt"].read_text(encoding="utf-8")) visual = json.loads(paths[f"{stage}_qa_receipt"].read_text(encoding="utf-8")) video_hash = sha256(paths[stage]) if render.get("status") != "completed" or render.get("verified") is not True or render.get("source_canon_sha256") != canon or render.get("artifact_sha256") != video_hash: raise RuntimeError(f"{stage} render receipt absent, failed, or stale") if visual.get("status") != "completed" or visual.get("verified") is not True or visual.get("source_canon_sha256") != canon or visual.get("video_sha256") != video_hash: raise RuntimeError(f"{stage} visual QA absent, failed, or stale") inputs[stage] = { "path": str(paths[stage]), "sha256": video_hash, "render_receipt": {"path": str(paths[f"{stage}_receipt"]), "sha256": sha256(paths[f"{stage}_receipt"])}, "visual_receipt": {"path": str(paths[f"{stage}_qa_receipt"]), "sha256": sha256(paths[f"{stage}_qa_receipt"])}, } payload = { "intro_video_path": str(paths["intro"]), "footage_video_path": str(paths["footage"]), "output_video_path": str(paths["final_master"]), "fps": 30, "width": 1920, "height": 1080, "video_encoder": "h264_nvenc", "preset": "p4", "cq": 21, } receipt_path = paths["final_receipt"] job_id = None if receipt_path.exists(): prior = json.loads(receipt_path.read_text(encoding="utf-8")) if prior.get("status") == "completed" and prior.get("verified") is True: raise RuntimeError("final already completed and requires audit") if prior.get("status") in {"submitted", "queued", "running"}: if prior.get("request") != payload: raise RuntimeError("existing final job payload differs") job_id = prior.get("job_id") elif paths["final_master"].exists(): raise RuntimeError("final output exists without receipt") if job_id is None: response = request_json(f"{BASE}/v1/huyenan-render-final/jobs", payload) job_id = response.get("job_id") or response.get("id") if not isinstance(job_id, str) or not job_id: raise RuntimeError(f"submit response lacks job id: {response}") atomic_json(receipt_path, { "status": response.get("status", "submitted"), "verified": False, "source_canon_sha256": canon, "job_id": job_id, "request": payload, "input_artifacts": inputs, "submit_response": response, "submitted_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), }) while True: job = request_json(f"{BASE}/v1/huyenan-render-final/jobs/{job_id}") status = job.get("status") if status in {"submitted", "queued", "running"}: current = json.loads(receipt_path.read_text(encoding="utf-8")) current.update({"status": status, "verified": False, "last_job_response": job, "updated_at": datetime.datetime.now(datetime.timezone.utc).isoformat()}) atomic_json(receipt_path, current) time.sleep(5) continue if status != "completed": current = json.loads(receipt_path.read_text(encoding="utf-8")) current.update({"status": status or "failed", "verified": False, "terminal_job_response": job, "failed_at": datetime.datetime.now(datetime.timezone.utc).isoformat()}) atomic_json(receipt_path, current) raise RuntimeError(f"final job terminal failure: {status}") break verification = job.get("verification") or {} output = Path(job.get("output", "")) final_probe = probe(paths["final_master"]) intro_probe = probe(paths["intro"]) footage_probe = probe(paths["footage"]) streams = final_probe.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"] expected_duration = duration(intro_probe) + duration(footage_probe) actual_duration = duration(final_probe) checks = { "endpoint_verified": verification.get("verified") is True, "output_exact": output == paths["final_master"], "output_exists": paths["final_master"].is_file() and paths["final_master"].stat().st_size > 0, "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": len(videos) == 1 and videos[0].get("width") == 1920 and videos[0].get("height") == 1080, "audio_48000": len(audios) == 1 and audios[0].get("sample_rate") == "48000", "duration_sum": abs(actual_duration - expected_duration) <= 1 / 30 + 0.005, "main_audio_not_sent": "main_audio_path" not in payload, } verified = all(checks.values()) receipt = { "status": "completed" if verified else "failed", "verified": verified, "source_canon_sha256": canon, "stage": "final", "job_id": job_id, "request": payload, "input_artifacts": inputs, "terminal_job_response": job, "output_path": str(paths["final_master"]), "artifact_sha256": sha256(paths["final_master"]), "bytes": paths["final_master"].stat().st_size, "duration_seconds": actual_duration, "duration_basis": {"intro": duration(intro_probe), "footage": duration(footage_probe), "expected_sum": expected_duration, "output": actual_duration}, "endpoint_verification": verification, "independent_probe": final_probe, "checks": {key: {"verified": value, "value": value} for key, value in checks.items()}, "completed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } atomic_json(receipt_path, receipt) print(json.dumps({"status": receipt["status"], "verified": verified, "job_id": job_id, "duration_seconds": actual_duration, "artifact_sha256": receipt["artifact_sha256"], "bytes": receipt["bytes"], "receipt": str(receipt_path)}, ensure_ascii=False)) return 0 if verified else 1 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)