#!/usr/bin/env python3 from __future__ import annotations import datetime import hashlib import json import os import subprocess import urllib.parse import urllib.request from pathlib import Path ROOT = Path("/data/video-pipeline/HaTramAudio/project/014-Nguoi-Ve-Ban-Do-Mua-Dong") PROJECT_ID = "014-Nguoi-Ve-Ban-Do-Mua-Dong" RUN_ID = "run-20260720T152357Z-c85a2128" OWNER = "zoro" ENDPOINT = "http://192.168.40.32:7861/voice/ngoc-huyen-clone" TEXT = "Các bạn đang nghe truyện được phát từ Hạ Trâm Audio, chúc các bạn có những giây phút nghe truyện vui vẻ. Hãy ủng hộ chúng tôi bằng cách like video và đăng ký kênh." POSTER = ROOT / "image" / "normalized" / "intro-poster-1920x1080.png" ARTWORK_RECEIPT = ROOT / "log" / "poster.json" PROMOTION = ROOT / "script" / "promotion-report.json" VOICE = ROOT / "audio" / "intro-voice.wav" SILENCE = ROOT / "audio" / "intro-silence-2s.wav" FULL = ROOT / "audio" / "intro-full.wav" OUTPUT = ROOT / "output" / "intro" / "intro.mp4" RECEIPT = ROOT / "log" / "intro-render.json" def now() -> str: return datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z") def sha(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for block in iter(lambda: f.read(8 * 1024 * 1024), b""): h.update(block) return h.hexdigest() def guard() -> None: lock = json.loads((ROOT / ".ownership-lock.json").read_text(encoding="utf-8")) for key, value in {"project_id": PROJECT_ID, "run_id": RUN_ID, "owner": OWNER, "status": "main_session_pipeline"}.items(): if lock.get(key) != value: raise RuntimeError(f"ownership mismatch: {key}") def probe(path: Path) -> dict: return json.loads(subprocess.check_output([ "ffprobe", "-v", "error", "-show_entries", "format=duration,size:stream=index,codec_type,codec_name,width,height,sample_rate,channels", "-of", "json", str(path), ])) def duration(path: Path) -> float: return float(probe(path)["format"]["duration"]) def atomic_json(path: Path, value: dict) -> None: guard() temp = path.with_name("." + path.name + ".tmp") temp.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") os.replace(temp, path) def synthesize() -> None: payload = urllib.parse.urlencode({"text": TEXT, "style": "doc_truyen", "speed": "1.0", "denoise": "true"}).encode() request = urllib.request.Request(ENDPOINT, data=payload, headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") temp = VOICE.with_name(".intro-voice.tmp.wav") with urllib.request.urlopen(request, timeout=900) as response, temp.open("wb") as out: while block := response.read(1024 * 1024): out.write(block) out.flush() os.fsync(out.fileno()) with temp.open("rb") as f: header = f.read(12) if header[:4] != b"RIFF" or header[8:12] != b"WAVE": raise RuntimeError("intro TTS is not WAV") guard() os.replace(temp, VOICE) def main() -> None: guard() promotion = json.loads(PROMOTION.read_text(encoding="utf-8")) art = json.loads(ARTWORK_RECEIPT.read_text(encoding="utf-8")) canon_hash = promotion.get("canon_sha256") or promotion.get("source_canon_sha256") if promotion.get("status") not in {"passed", "completed", "completed_content_only"} or not promotion.get("verified") or not art.get("verified"): raise RuntimeError("promotion/artwork gate is not verified") if art.get("source_canon_sha256") != canon_hash: raise RuntimeError("artwork canon hash mismatch") poster_hash = sha(POSTER) expected_poster_hash = art.get("normalized_intro_poster_sha256") or art.get("intro_poster_sha256") if poster_hash != expected_poster_hash: raise RuntimeError("poster hash mismatch") for path in [VOICE.parent, OUTPUT.parent, RECEIPT.parent]: path.mkdir(parents=True, exist_ok=True) synthesize() voice_probe = probe(VOICE) streams = voice_probe.get("streams", []) if len(streams) != 1 or streams[0].get("codec_name") != "pcm_s16le" or streams[0].get("sample_rate") != "48000" or streams[0].get("channels") != 1: raise RuntimeError("intro voice format mismatch") subprocess.check_call([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "lavfi", "-i", "anullsrc=r=48000:cl=mono", "-t", "2.000", "-c:a", "pcm_s16le", str(SILENCE), ]) concat = ROOT / "work" / "intro.ffconcat" concat.write_text("ffconcat version 1.0\n" + f"file '{VOICE}'\nfile '{SILENCE}'\n", encoding="utf-8") temp_audio = FULL.with_name(".intro-full.tmp.wav") subprocess.check_call([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "concat", "-safe", "0", "-i", str(concat), "-c", "copy", str(temp_audio), ]) guard() os.replace(temp_audio, FULL) full_duration = duration(FULL) temp_video = OUTPUT.with_name(".intro.tmp.mp4") subprocess.check_call([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-loop", "1", "-i", str(POSTER), "-i", str(FULL), "-t", f"{full_duration:.6f}", "-c:v", "libx264", "-preset", "medium", "-crf", "18", "-r", "30", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2", "-movflags", "+faststart", str(temp_video), ]) out_probe = probe(temp_video) out_duration = float(out_probe["format"]["duration"]) if abs(out_duration - full_duration) > 1 / 30 + 0.01: raise RuntimeError("intro duration mismatch") guard() os.replace(temp_video, OUTPUT) receipt = { "schema_version": 1, "project_id": PROJECT_ID, "run_id": RUN_ID, "status": "completed", "verified": True, "source_canon_sha256": canon_hash, "source_poster_sha256": poster_hash, "intro_text": TEXT, "intro_text_sha256": hashlib.sha256(TEXT.encode("utf-8")).hexdigest(), "voice_path": str(VOICE), "voice_sha256": sha(VOICE), "silence_path": str(SILENCE), "silence_duration_seconds": duration(SILENCE), "audio_path": str(FULL), "audio_sha256": sha(FULL), "audio_duration_seconds": full_duration, "artifact_path": str(OUTPUT), "artifact_sha256": sha(OUTPUT), "artifact_bytes": OUTPUT.stat().st_size, "duration_seconds": out_duration, "duration_delta_seconds": abs(out_duration - full_duration), "probe": probe(OUTPUT), "completed_at": now(), } atomic_json(RECEIPT, receipt) print(json.dumps({"status": "completed", "verified": True, "duration": out_duration, "artifact_sha256": receipt["artifact_sha256"]}, ensure_ascii=False)) if __name__ == "__main__": main()