#!/usr/bin/env python3 import datetime import hashlib import json import os import subprocess import urllib.request import uuid from pathlib import Path ROOT = Path(__file__).resolve().parents[1] BASE = "http://192.168.1.104:8021" CREATE = "/stickman-render-gpu" PROJECT_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 probe(path): result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration,size", "-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 request_json(path, payload=None): data = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8") request = urllib.request.Request( BASE + path, data=data, headers={"Content-Type": "application/json"} if data else {}, method="POST" if data else "GET", ) with urllib.request.urlopen(request, timeout=600) as response: if response.status // 100 != 2: raise RuntimeError(f"HTTP {response.status}: {path}") return json.load(response) 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 main(): manifest = json.loads(PROJECT_MANIFEST.read_text(encoding="utf-8")) active = manifest["active_paths"] poster = ROOT / active["intro_poster"] audio = ROOT / active["intro_audio"] audio_receipt_path = ROOT / active["intro_audio_receipt"] layout_receipt_path = ROOT / active["layout_receipt"] output = ROOT / active["intro"] receipt_path = ROOT / active["intro_receipt"] if output.exists() or receipt_path.exists(): raise RuntimeError("intro output/receipt must be virgin") audio_receipt = json.loads(audio_receipt_path.read_text(encoding="utf-8")) layout_receipt = json.loads(layout_receipt_path.read_text(encoding="utf-8")) if not (audio_receipt.get("verified") is True and layout_receipt.get("verified") is True): raise RuntimeError("intro upstream gate is not verified") if sha256(audio) != audio_receipt["intro_full_sha256"]: raise RuntimeError("intro audio hash drift") layout_intro = layout_receipt["artifacts"]["intro_normalized"] if sha256(poster) != layout_intro["sha256"]: raise RuntimeError("intro poster hash drift") audio_probe = probe(audio) duration = float(audio_probe["format"]["duration"]) health = request_json("/health") if health.get("status") != "ok" or health.get("nvenc_available") is not True: raise RuntimeError("intro endpoint health/NVENC failed") if health.get("encoder") != "h264_nvenc": raise RuntimeError("intro endpoint encoder contract failed") if health.get("video_root") != "/data/video-pipeline": raise RuntimeError("intro endpoint video root mismatch") output.parent.mkdir(parents=True, exist_ok=True) payload = { "job_id": f"gac-mai-004-intro-{uuid.uuid4().hex[:12]}", "scenes": [{"image_path": str(poster), "duration": duration}], "audio_path": str(audio), "subtitle_path": None, "output_path": str(output), "width": 1920, "height": 1080, "fps": 30, "crf": 20, "preset": "p4", "burn_subtitles": False, "ken_burns": False, "motion_preset": "none", "transition": "cut", "transition_duration": 0.0, "fade_in": 0.0, "fade_out": 0.0, "slice_audio": False, "slice_subtitles": False, } response = request_json(CREATE, payload) if not output.is_file() or output.stat().st_size <= 0: raise RuntimeError("intro endpoint returned without closed output") info = probe(output) videos = [row for row in info.get("streams", []) if row.get("codec_type") == "video"] audios = [row for row in info.get("streams", []) if row.get("codec_type") == "audio"] output_duration = float(info["format"]["duration"]) checks = { "endpoint_response_terminal": response.get("status") in (None, "completed", "success", "ok"), "output_path_exact": response.get("output_path", response.get("output", str(output))) == str(output), "one_h264_video": len(videos) == 1 and videos[0].get("codec_name") == "h264", "one_audio": len(audios) == 1, "resolution": len(videos) == 1 and (videos[0].get("width"), videos[0].get("height")) == (1920, 1080), "duration_one_frame": abs(output_duration - duration) <= 1 / 30 + 0.005, "subtitles_disabled": payload["subtitle_path"] is None and payload["burn_subtitles"] is False, "server_encoder": health.get("encoder") == "h264_nvenc", } if not all(checks.values()): raise RuntimeError(f"intro verification failed: {checks}") now = datetime.datetime.now(datetime.timezone.utc).isoformat() receipt = { "status": "completed", "verified": True, "stage": "intro", "service": health.get("service"), "base_url": BASE, "create_endpoint": CREATE, "job_id": payload["job_id"], "source_canon_sha256": manifest["canon_sha256"], "poster_path": str(poster.relative_to(ROOT)), "poster_sha256": sha256(poster), "intro_audio_path": str(audio.relative_to(ROOT)), "intro_audio_sha256": sha256(audio), "input_duration_seconds": duration, "output_path": str(output.relative_to(ROOT)), "artifact_sha256": sha256(output), "artifact_bytes": output.stat().st_size, "duration_seconds": output_duration, "request": payload, "terminal_response": response, "independent_probe": info, "checks": checks, "completed_at": now, } atomic_json(receipt_path, receipt) manifest["steps"]["intro"] = "completed" manifest["intro"] = {"status": "completed", "verified": True, "receipt": str(receipt_path.relative_to(ROOT)), "artifact_sha256": receipt["artifact_sha256"], "duration_seconds": output_duration, "completed_at": now} manifest["updated_at"] = now atomic_json(PROJECT_MANIFEST, manifest) print(json.dumps({"status": "completed", "job_id": payload["job_id"], "duration_seconds": output_duration, "sha256": receipt["artifact_sha256"]}, ensure_ascii=False)) if __name__ == "__main__": main()