#!/usr/bin/env python3 import argparse import datetime import hashlib import json import os import subprocess import time import urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parents[1] MANIFEST = ROOT / "script/project-manifest.json" CONFIG = { "final": { "base": "http://192.168.1.104:8023", "create": "/v1/huyenan-render-final/jobs", "receipt": "final_receipt", "output": "final_master", "upstream": ["intro_receipt", "footage_receipt"], }, "transcode": { "base": "http://192.168.1.104:8024", "create": "/v1/huyenan-transcode-upload/jobs", "receipt": "transcode_receipt", "output": "final_upload", "upstream": ["final_receipt", "final_qa_receipt"], }, } 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.resolve().relative_to(ROOT.resolve()) return path def atomic_json(path, value): path.parent.mkdir(parents=True, exist_ok=True) 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 api(url, payload=None, timeout=60): data = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8") 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=timeout) as response: return json.load(response) def probe(path): result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration,size,format_name", "-show_entries", "stream=index,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 payload_sha(payload): return hashlib.sha256(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() def payload_for(stage, project, output): if stage == "final": return { "intro_video_path": str(resolve(project, "intro")), "footage_video_path": str(resolve(project, "footage")), "output_video_path": str(output), "fps": 30, "width": 1920, "height": 1080, "video_encoder": "h264_nvenc", "preset": "p4", "cq": 21, } return { "input_video_path": str(resolve(project, "final_master")), "output_video_path": str(output), "target_output_bytes": 800000000, "max_output_bytes": 1000000000, "width": 1920, "height": 1080, "fps": 30, "audio_bitrate_kbps": 96, "overwrite": False, } def verify_upstream(project, keys, canon): rows = [] for key in keys: path = resolve(project, key) value = json.loads(path.read_text(encoding="utf-8")) if value.get("status") != "completed" or value.get("verified") is not True or value.get("source_canon_sha256") != canon: raise RuntimeError(f"upstream lineage failed: {key}") artifact_rel = value.get("output_path") artifact = ROOT / artifact_rel if artifact_rel and not Path(artifact_rel).is_absolute() else Path(artifact_rel or "") if not artifact.is_file() or sha256(artifact) != value.get("artifact_sha256"): raise RuntimeError(f"upstream artifact drift: {key}") rows.append({"key": key, "path": str(path), "sha256": sha256(path)}) return rows def main(): parser = argparse.ArgumentParser() parser.add_argument("stage", choices=CONFIG) args = parser.parse_args() stage = args.stage cfg = CONFIG[stage] project = json.loads(MANIFEST.read_text(encoding="utf-8")) canon = project["canon_sha256"] upstream = verify_upstream(project, cfg["upstream"], canon) output = resolve(project, cfg["output"]) receipt_path = resolve(project, cfg["receipt"]) health = api(cfg["base"] + "/health") if health.get("status") != "ok" or health.get("h264_nvenc") is not True or health.get("video_root") != "/data/video-pipeline": raise RuntimeError(f"{stage} endpoint health/path policy failed") payload = payload_for(stage, project, output) request_sha = payload_sha(payload) job_id = None if receipt_path.exists(): state = json.loads(receipt_path.read_text(encoding="utf-8")) if state.get("status") in {"submitted", "queued", "running"} and state.get("source_canon_sha256") == canon and state.get("request_sha256") == request_sha and state.get("request") == payload: job_id = str(state["job_id"]) else: raise RuntimeError(f"existing {stage} receipt is not safely resumable") else: if output.exists(): raise RuntimeError(f"{stage} output exists without live receipt") created = api(cfg["base"] + cfg["create"], payload) a, b = created.get("job_id"), created.get("id") if a and b and str(a) != str(b): raise RuntimeError("conflicting job identifiers") job_id = str(a or b or "") if not job_id: raise RuntimeError("create response missing job id") atomic_json(receipt_path, { "status": "submitted", "verified": False, "stage": stage, "source_canon_sha256": canon, "job_id": job_id, "status_url": cfg["base"] + cfg["create"] + "/" + job_id, "request": payload, "request_sha256": request_sha, "health": health, "input_receipts": upstream, "created_response": created, "submitted_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), }) status_url = cfg["base"] + cfg["create"] + "/" + job_id while True: job = api(status_url) status = str(job.get("status", "unknown")) if status in {"completed", "failed", "cancelled"}: break state = json.loads(receipt_path.read_text(encoding="utf-8")) state.update({"status": status, "latest_response": job, "updated_at": datetime.datetime.now(datetime.timezone.utc).isoformat()}) atomic_json(receipt_path, state) time.sleep(5) if status != "completed": state = json.loads(receipt_path.read_text(encoding="utf-8")) state.update({"status": status, "latest_response": job, "failed_at": datetime.datetime.now(datetime.timezone.utc).isoformat()}) atomic_json(receipt_path, state) raise RuntimeError(f"{stage} terminal failure: {status}") verification = job.get("verification") or {} if verification.get("verified") is not True or not output.is_file() or output.stat().st_size <= 0: raise RuntimeError(f"{stage} terminal verification/output failed") 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"]) response_output = str(job.get("output") or job.get("output_path") or job.get("output_video_path") or "") checks = { "endpoint_completed": status == "completed", "endpoint_verified": verification.get("verified") is True, "output_path_exact": response_output == str(output), "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"), videos[0].get("height")) == (1920, 1080), "positive_duration": duration > 0, } basis = {"output": duration} if stage == "final": intro_duration = float(probe(resolve(project, "intro"))["format"]["duration"]) footage_duration = float(probe(resolve(project, "footage"))["format"]["duration"]) basis.update({"intro": intro_duration, "footage": footage_duration, "expected_sum": intro_duration + footage_duration}) checks["duration_matches_intro_plus_footage"] = abs(duration - intro_duration - footage_duration) <= 0.08 checks["narration_not_remapped"] = "main_audio_path" not in payload else: source_duration = float(probe(resolve(project, "final_master"))["format"]["duration"]) basis["source_final"] = source_duration checks.update({ "duration_matches_master": abs(duration - source_duration) <= 0.08, "under_one_billion_bytes": output.stat().st_size < 1000000000, "endpoint_under_one_gb": verification.get("under_one_gb") is True, }) if not all(checks.values()): raise RuntimeError(f"{stage} integrity verification failed: {checks}") now = datetime.datetime.now(datetime.timezone.utc).isoformat() receipt = { "status": "completed", "verified": True, "stage": stage, "source_canon_sha256": canon, "job_id": job_id, "request": payload, "request_sha256": request_sha, "input_receipts": upstream, "output_path": str(output.relative_to(ROOT)), "artifact_sha256": sha256(output), "artifact_bytes": output.stat().st_size, "duration_seconds": duration, "duration_basis": basis, "endpoint_verification": verification, "independent_probe": info, "checks": checks, "latest_response": job, "hash_method": "sha256 streaming 8 MiB", "completed_at": now, } atomic_json(receipt_path, receipt) latest = json.loads(MANIFEST.read_text(encoding="utf-8")) step = "final_render" if stage == "final" else "transcode" latest["steps"][step] = "completed" latest[stage] = {"status": "completed", "verified": True, "job_id": job_id, "receipt": str(receipt_path.relative_to(ROOT)), "artifact_sha256": receipt["artifact_sha256"], "artifact_bytes": receipt["artifact_bytes"], "duration_seconds": duration, "completed_at": now} latest["updated_at"] = now atomic_json(MANIFEST, latest) print(json.dumps({"status": "completed", "verified": True, "stage": stage, "job_id": job_id, "bytes": output.stat().st_size, "duration_seconds": duration, "sha256": receipt["artifact_sha256"]}, ensure_ascii=False)) if __name__ == "__main__": main()