#!/usr/bin/env python3 from __future__ import annotations import argparse 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] PROJECT_MANIFEST = ROOT / "script/project-manifest.json" STAGES = { "footage": {"base": "http://192.168.1.104:8022", "create": "/v1/footage-render-huyenan/jobs", "receipt_key": "footage_receipt", "output_key": "footage", "upstream_keys": ["tts_receipt", "layout_receipt"]}, "final": {"base": "http://192.168.1.104:8023", "create": "/v1/huyenan-render-final/jobs", "receipt_key": "final_receipt", "output_key": "final_master", "upstream_keys": ["intro_receipt", "intro_qa_receipt", "footage_receipt"]}, "transcode": {"base": "http://192.168.1.104:8024", "create": "/v1/huyenan-transcode-upload/jobs", "receipt_key": "transcode_receipt", "output_key": "final_upload", "upstream_keys": ["final_receipt", "final_qa_receipt"]}, } 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 get_json(url: str, timeout: int = 30) -> dict: with urllib.request.urlopen(url, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) def post_json(url: str, payload: dict, timeout: int = 60) -> dict: request = urllib.request.Request(url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST") with urllib.request.urlopen(request, timeout=timeout) 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,format_name:stream=index,codec_name,codec_type,width,height,sample_rate,channels", "-of", "json", str(path), ], text=True)) def request_payload(stage: str, project: dict, output: Path) -> dict: if stage == "footage": seed_material = str(project.get("footage_seed_material") or project.get("project_id", "")) seed = int(hashlib.sha256(seed_material.encode("utf-8")).hexdigest()[:12], 16) footage_dir = str(resolve_active(project, "footage_source_view")) if (project.get("active_paths") or {}).get("footage_source_view") else "/data/video-pipeline/GacMaiAudio/footage" return { "layout": str(resolve_active(project, "layout")), "footage_dir": footage_dir, "pattern": "*.mp4", "recursive": True, "audio": str(resolve_active(project, "tts_audio")), "output": str(output), "seed": seed, "overwrite": False, "width": 1920, "height": 1080, "fps": 30, "center_x": 656, "center_y": 0, "center_width": 608, "center_height": 1080, "video_encoder": "h264_nvenc", "preset": "p4", "cq": 21, } if stage == "final": return { "intro_video_path": str(resolve_active(project, "intro")), "footage_video_path": str(resolve_active(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_active(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: dict, keys: list[str], canon_hash: str) -> list[dict]: values = [] for key in keys: path = resolve_active(project, key) if not path.exists(): raise RuntimeError(f"upstream receipt missing: {key} -> {path}") value = json.loads(path.read_text(encoding="utf-8")) if value.get("status") != "completed" or value.get("verified") is not True: raise RuntimeError(f"upstream receipt not completed/verified: {key}") if value.get("source_canon_sha256") != canon_hash: raise RuntimeError(f"upstream canon hash mismatch: {key}") values.append({"key": key, "path": str(path), "sha256": sha256(path)}) return values def payload_hash(payload: dict) -> str: raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(raw).hexdigest() def current_job(receipt: Path, canon_hash: str, payload: dict) -> str | None: if not receipt.exists(): return None data = json.loads(receipt.read_text(encoding="utf-8")) if data.get("source_canon_sha256") != canon_hash: raise RuntimeError("existing render receipt is stale for canon") if data.get("status") in {"submitted", "queued", "running"} and data.get("job_id"): if data.get("request_sha256") != payload_hash(payload) or data.get("request") != payload: raise RuntimeError("live render receipt payload lineage mismatch") return str(data["job_id"]) raise RuntimeError("terminal render receipt already exists and requires audit/invalidation") def independent_verify(stage: str, project: dict, output: Path, job: dict, canon_hash: str, upstream: list[dict]) -> dict: if not output.exists() or output.stat().st_size <= 0: raise RuntimeError("render output missing or empty") info = probe(output) streams = info.get("streams", []) video = [item for item in streams if item.get("codec_type") == "video"] audio = [item for item in streams if item.get("codec_type") == "audio"] duration = float(info["format"]["duration"]) checks = { "endpoint_status_completed": job.get("status") == "completed", "endpoint_verified": (job.get("verification") or {}).get("verified") is True, "output_path_exact": str(job.get("output") or job.get("output_path") or job.get("output_video_path")) == str(output), "one_video": len(video) == 1, "one_audio": len(audio) == 1, "resolution": len(video) == 1 and (int(video[0].get("width", 0)), int(video[0].get("height", 0))) == (1920, 1080), "video_h264": len(video) == 1 and video[0].get("codec_name") == "h264", "audio_aac": len(audio) == 1 and audio[0].get("codec_name") == "aac", "positive_duration": duration > 0, } basis = {"output": duration} if stage == "footage": verification = job.get("verification") or {} source_duration = float(probe(resolve_active(project, "tts_audio"))["format"]["duration"]) basis["story_full"] = source_duration checks.update({"mirror_applied": verification.get("mirror_applied") is True, "source_audio_discarded": verification.get("source_audio_discarded") is True, "audio_status": verification.get("audio_status") in {"narration_mixed", "present", "ok"}, "duration_matches_story_audio": abs(duration - source_duration) <= 0.08}) elif stage == "final": intro_duration = float(probe(resolve_active(project, "intro"))["format"]["duration"]) footage_duration = float(probe(resolve_active(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 else: source_duration = float(probe(resolve_active(project, "final_master"))["format"]["duration"]) basis["source_final"] = source_duration checks.update({"under_one_gb": output.stat().st_size < 1_000_000_000, "endpoint_under_one_gb": (job.get("verification") or {}).get("under_one_gb") is True, "duration_matches_master": abs(duration - source_duration) <= 0.08}) verified = all(checks.values()) return {"status": "completed" if verified else "failed", "verified": verified, "source_canon_sha256": canon_hash, "stage": stage, "job_id": str(job.get("job_id") or job.get("id")), "input_receipts": upstream, "output_path": str(output), "artifact_sha256": sha256(output), "artifact_bytes": output.stat().st_size, "duration_seconds": duration, "duration_basis": basis, "endpoint_verification": job.get("verification"), "independent_probe": info, "checks": {k: {"verified": v, "value": v} for k, v in checks.items()}, "latest_response": job, "completed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "hash_method": "sha256 streaming 8 MiB", "blockers": [k for k, v in checks.items() if not v]} def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("stage", choices=STAGES) parser.add_argument("--max-wait", type=int, default=14400) args = parser.parse_args() stage = args.stage cfg = STAGES[stage] project = json.loads(PROJECT_MANIFEST.read_text(encoding="utf-8")) promotion_path = resolve_active(project, "promotion_receipt") promotion = json.loads(promotion_path.read_text(encoding="utf-8")) if promotion.get("status") != "completed" or promotion.get("verified") is not True: raise RuntimeError("promotion is absent or not verified") canon_hash = promotion["canon_sha256"] receipt = resolve_active(project, cfg["receipt_key"]) output = resolve_active(project, cfg["output_key"]) upstream = verify_upstream(project, cfg["upstream_keys"], canon_hash) health = get_json(cfg["base"] + "/health") if health.get("status") != "ok" or health.get("h264_nvenc") is not True: raise RuntimeError(f"worker health/NVENC gate failed: {health}") payload = request_payload(stage, project, output) request_sha256 = payload_hash(payload) job_id = current_job(receipt, canon_hash, payload) if job_id is None: if output.exists(): raise RuntimeError(f"output exists without a live audited job: {output}") created = post_json(cfg["base"] + cfg["create"], payload) raw_job_id = created.get("job_id") raw_id = created.get("id") if raw_job_id and raw_id and str(raw_job_id) != str(raw_id): raise RuntimeError(f"create response contains conflicting job identifiers: {created}") job_id = str(raw_job_id or raw_id or "") if not job_id: raise RuntimeError(f"create response has no job id: {created}") atomic_json(receipt, {"status": "submitted", "verified": False, "stage": stage, "source_canon_sha256": canon_hash, "job_id": job_id, "status_url": cfg["base"] + cfg["create"] + "/" + job_id, "request": payload, "request_sha256": request_sha256, "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 deadline = time.monotonic() + args.max_wait while True: job = get_json(status_url) status = str(job.get("status", "unknown")) if status in {"completed", "failed", "cancelled"}: break atomic_json(receipt, {"status": status, "verified": False, "stage": stage, "source_canon_sha256": canon_hash, "job_id": job_id, "status_url": status_url, "request": payload, "request_sha256": request_sha256, "health": health, "input_receipts": upstream, "latest_response": job, "updated_at": datetime.datetime.now(datetime.timezone.utc).isoformat()}) if time.monotonic() >= deadline: print(json.dumps({"status": status, "verified": False, "job_id": job_id, "resume_required": True, "receipt": str(receipt)}, ensure_ascii=False)) return 2 time.sleep(5) if status != "completed": atomic_json(receipt, {"status": status, "verified": False, "stage": stage, "source_canon_sha256": canon_hash, "job_id": job_id, "latest_response": job, "failed_at": datetime.datetime.now(datetime.timezone.utc).isoformat()}) raise RuntimeError(f"{stage} job terminal failure: {status}") result = independent_verify(stage, project, output, job, canon_hash, upstream) atomic_json(receipt, result) print(json.dumps({"status": result["status"], "verified": result["verified"], "stage": stage, "job_id": job_id, "output": str(output), "receipt": str(receipt), "blockers": result["blockers"]}, ensure_ascii=False)) return 0 if result["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)