#!/usr/bin/env python3 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" BASE = "http://192.168.1.104:8022" CREATE = "/v1/footage-render-huyenan/jobs" 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 get_json(url, timeout=30): with urllib.request.urlopen(url, timeout=timeout) as response: return json.load(response) def post_json(url, payload, timeout=60): 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.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_hash(payload): raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(raw).hexdigest() def main(): project = json.loads(MANIFEST.read_text(encoding="utf-8")) canon = project["canon_sha256"] promotion_path = resolve(project, "promotion_receipt") tts_receipt_path = resolve(project, "tts_receipt") layout_receipt_path = resolve(project, "layout_receipt") receipt_path = resolve(project, "footage_receipt") output = resolve(project, "footage") layout = resolve(project, "layout") audio = resolve(project, "tts_audio") promotion = json.loads(promotion_path.read_text(encoding="utf-8")) tts = json.loads(tts_receipt_path.read_text(encoding="utf-8")) layout_receipt = json.loads(layout_receipt_path.read_text(encoding="utf-8")) if promotion.get("verified") is not True or promotion.get("source_canon_sha256") != canon: raise RuntimeError("promotion lineage failed") if tts.get("verified") is not True or tts.get("source_canon_sha256") != canon or sha256(audio) != tts.get("artifact_sha256"): raise RuntimeError("TTS lineage failed") if layout_receipt.get("verified") is not True or layout_receipt.get("source_canon_sha256") != canon: raise RuntimeError("layout lineage failed") layout_row = layout_receipt.get("artifacts", {}).get("layout", {}) if sha256(layout) != layout_row.get("sha256"): raise RuntimeError("layout artifact drift") health = get_json(BASE + "/health") if health.get("status") != "ok" or health.get("h264_nvenc") is not True: raise RuntimeError("footage health/NVENC failed") if health.get("video_root") != "/data/video-pipeline" or health.get("default_footage_dir") != "/data/video-pipeline/GacMaiAudio/footage" or health.get("output_policy") != "request_path_under_video_root": raise RuntimeError("footage endpoint path policy failed") seed = int(hashlib.sha256(project["project_id"].encode("utf-8")).hexdigest()[:12], 16) payload = { "layout": str(layout), "footage_dir": "/data/video-pipeline/GacMaiAudio/footage", "pattern": "*.mp4", "recursive": True, "audio": str(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, } request_sha = payload_hash(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("existing footage receipt is not safely resumable") else: if output.exists(): raise RuntimeError("footage output exists without active receipt") created = post_json(BASE + 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("conflicting footage job identifiers") job_id = str(raw_job_id or raw_id or "") if not job_id: raise RuntimeError("footage create response missing job id") atomic_json(receipt_path, { "status": "submitted", "verified": False, "stage": "footage", "source_canon_sha256": canon, "job_id": job_id, "status_url": BASE + CREATE + "/" + job_id, "request": payload, "request_sha256": request_sha, "health": health, "input_receipts": [ {"key": "tts_receipt", "path": str(tts_receipt_path), "sha256": sha256(tts_receipt_path)}, {"key": "layout_receipt", "path": str(layout_receipt_path), "sha256": sha256(layout_receipt_path)}, ], "created_response": created, "submitted_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), }) status_url = BASE + CREATE + "/" + job_id while True: job = get_json(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"footage terminal failure: {status}") verification = job.get("verification") or {} if verification.get("verified") is not True: raise RuntimeError("footage endpoint verification failed") if not output.is_file() or output.stat().st_size <= 0: raise RuntimeError("footage terminal output missing") 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"]) source_duration = float(probe(audio)["format"]["duration"]) response_output = str(job.get("output") or job.get("output_path") or "") horizontal_mirror = job.get("horizontal_mirror") or verification.get("horizontal_mirror") 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), "duration_matches_narration": abs(duration - source_duration) <= 0.08, "mirror_applied": verification.get("mirror_applied") is True, "horizontal_mirror_hflip": horizontal_mirror == "hflip", "source_audio_discarded": verification.get("source_audio_discarded") is True, "narration_audio": verification.get("audio_status") in {"narration_mixed", "narration_attached"}, } if not all(checks.values()): raise RuntimeError(f"footage integrity verification failed: {checks}") now = datetime.datetime.now(datetime.timezone.utc).isoformat() previous = json.loads(receipt_path.read_text(encoding="utf-8")) receipt = { "status": "completed", "verified": True, "stage": "footage", "source_canon_sha256": canon, "job_id": job_id, "request": payload, "request_sha256": request_sha, "input_receipts": previous["input_receipts"], "output_path": str(output.relative_to(ROOT)), "artifact_sha256": sha256(output), "artifact_bytes": output.stat().st_size, "duration_seconds": duration, "duration_basis": {"output": duration, "story_full": source_duration}, "endpoint_verification": verification, "horizontal_mirror": "hflip", "mirror_applied": True, "source_audio_discarded": True, "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")) latest["steps"]["footage"] = "completed" latest["footage"] = {"status": "completed", "verified": True, "job_id": job_id, "receipt": str(receipt_path.relative_to(ROOT)), "artifact_sha256": receipt["artifact_sha256"], "duration_seconds": duration, "mirror_applied": True, "horizontal_mirror": "hflip", "source_audio_discarded": True, "completed_at": now} latest["updated_at"] = now atomic_json(MANIFEST, latest) print(json.dumps({"status": "completed", "verified": True, "job_id": job_id, "duration_seconds": duration, "sha256": receipt["artifact_sha256"]}, ensure_ascii=False)) if __name__ == "__main__": main()