#!/usr/bin/env python3 from __future__ import annotations import base64 import datetime import hashlib import json import os import subprocess import sys import tempfile import urllib.parse import urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parents[1] PROJECT_MANIFEST = ROOT / "script/project-manifest.json" ENDPOINT = "http://192.168.40.32:7861/voice/ngoc-huyen-clone" BRAND = "Gác Mái Audio" INTRO_TEMPLATE = "Các bạn đang nghe truyện được phát từ Gác Mái 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." 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_bytes(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) 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 atomic_json(path: Path, value: dict) -> None: atomic_bytes(path, (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8")) def find_audio(value: object) -> bytes | None: if isinstance(value, dict): for key in ("audio_base64", "base64", "audio", "data"): item = value.get(key) if isinstance(item, str): raw = item.split(",", 1)[-1] if item.startswith("data:") else item try: decoded = base64.b64decode(raw, validate=True) except Exception: decoded = b"" if decoded.startswith(b"RIFF") and decoded[8:12] == b"WAVE": return decoded for item in value.values(): found = find_audio(item) if found: return found elif isinstance(value, list): for item in value: found = find_audio(item) if found: return found return None def probe(path: Path) -> dict: return json.loads(subprocess.check_output([ "ffprobe", "-v", "error", "-show_entries", "format=duration:stream=codec_name,codec_type,width,height,sample_rate,channels,r_frame_rate", "-of", "json", str(path), ], text=True)) def main() -> int: project = json.loads(PROJECT_MANIFEST.read_text(encoding="utf-8")) title = str(project.get("story_title") or "").strip() if not title: raise RuntimeError("project story_title is missing") text = INTRO_TEMPLATE.format(title=title) paths = {key: resolve_active(project, key) for key in ( "promotion_receipt", "layout_receipt", "intro_normalized", "intro_voice", "intro_silence", "intro_audio", "intro", "intro_receipt", )} promotion = json.loads(paths["promotion_receipt"].read_text(encoding="utf-8")) layout = json.loads(paths["layout_receipt"].read_text(encoding="utf-8")) canon_hash = promotion.get("canon_sha256") if promotion.get("status") != "completed" or promotion.get("verified") is not True: raise RuntimeError("promotion is not verified") if layout.get("status") != "completed" or layout.get("verified") is not True or layout.get("source_canon_sha256") != canon_hash: raise RuntimeError("layout receipt is absent, failed, or stale") normalized = layout.get("artifacts", {}).get(str(paths["intro_normalized"].relative_to(ROOT)), {}) if normalized.get("sha256") != sha256(paths["intro_normalized"]): raise RuntimeError("intro poster differs from verified layout artifact") existing = [str(paths[key]) for key in ("intro_voice", "intro_silence", "intro_audio", "intro", "intro_receipt") if paths[key].exists()] if existing: raise RuntimeError(f"intro production outputs already exist and require audit/invalidation: {existing}") for key in ("intro_voice", "intro_silence", "intro_audio", "intro", "intro_receipt"): paths[key].parent.mkdir(parents=True, exist_ok=True) payload = urllib.parse.urlencode({ "text": text, "style": "doc_truyen", "speed": "1.0", "denoise": "true", }).encode("utf-8") request = urllib.request.Request( ENDPOINT, data=payload, headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST", ) with urllib.request.urlopen(request, timeout=600) as response: body = response.read() http_status = response.status content_type = response.headers.get("Content-Type", "") audio = body if body.startswith(b"RIFF") and body[8:12] == b"WAVE" else None if audio is None: audio = find_audio(json.loads(body.decode("utf-8"))) if audio is None: raise RuntimeError("intro TTS response contains no WAV") atomic_bytes(paths["intro_voice"], audio) voice_probe = probe(paths["intro_voice"]) if not [item for item in voice_probe.get("streams", []) if item.get("codec_type") == "audio"]: raise RuntimeError("intro voice failed independent audio probe") subprocess.run([ "ffmpeg", "-v", "error", "-y", "-f", "lavfi", "-i", "anullsrc=r=48000:cl=mono", "-t", "2", "-c:a", "pcm_s16le", str(paths["intro_silence"]), ], check=True) subprocess.run([ "ffmpeg", "-v", "error", "-y", "-i", str(paths["intro_voice"]), "-i", str(paths["intro_silence"]), "-filter_complex", "[0:a][1:a]concat=n=2:v=0:a=1[a]", "-map", "[a]", "-ar", "48000", "-ac", "1", "-c:a", "pcm_s16le", str(paths["intro_audio"]), ], check=True) voice_duration = float(voice_probe["format"]["duration"]) silence_duration = float(probe(paths["intro_silence"])["format"]["duration"]) audio_duration = float(probe(paths["intro_audio"])["format"]["duration"]) if abs(silence_duration - 2.0) > 0.001 or abs(audio_duration - voice_duration - 2.0) > 0.02: raise RuntimeError("intro audio/silence duration verification failed") subprocess.run([ "ffmpeg", "-v", "error", "-y", "-loop", "1", "-i", str(paths["intro_normalized"]), "-i", str(paths["intro_audio"]), "-t", f"{audio_duration:.6f}", "-r", "30", "-c:v", "h264_nvenc", "-preset", "p4", "-cq", "21", "-pix_fmt", "yuv420p", "-c:a", "aac", "-ar", "48000", "-ac", "2", "-movflags", "+faststart", str(paths["intro"]), ], check=True) video_probe = probe(paths["intro"]) video_duration = float(video_probe["format"]["duration"]) streams = video_probe.get("streams", []) video_streams = [item for item in streams if item.get("codec_type") == "video"] audio_streams = [item for item in streams if item.get("codec_type") == "audio"] verified = ( http_status == 200 and abs(video_duration - audio_duration) <= 1 / 30 + 0.005 and len(video_streams) == 1 and len(audio_streams) == 1 and video_streams[0].get("codec_name") == "h264" and (int(video_streams[0].get("width", 0)), int(video_streams[0].get("height", 0))) == (1920, 1080) and audio_streams[0].get("codec_name") == "aac" ) receipt = { "status": "completed" if verified else "failed", "verified": verified, "source_canon_sha256": canon_hash, "layout_receipt": {"path": str(paths["layout_receipt"]), "sha256": sha256(paths["layout_receipt"])}, "input_path": str(paths["intro_normalized"]), "input_sha256": sha256(paths["intro_normalized"]), "output_path": str(paths["intro"]), "intro_text": text, "brand": BRAND, "endpoint": ENDPOINT, "style": "doc_truyen", "speed": 1.0, "denoise": True, "http_status": http_status, "response_content_type": content_type, "voice_path": str(paths["intro_voice"]), "voice_sha256": sha256(paths["intro_voice"]), "voice_duration_seconds": voice_duration, "silence_path": str(paths["intro_silence"]), "silence_sha256": sha256(paths["intro_silence"]), "silence_duration_seconds": silence_duration, "intro_full_path": str(paths["intro_audio"]), "intro_full_sha256": sha256(paths["intro_audio"]), "audio_duration_seconds": audio_duration, "artifact_sha256": sha256(paths["intro"]), "video_duration_seconds": video_duration, "duration_delta_seconds": abs(video_duration - audio_duration), "independent_probe": video_probe, "completed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "hash_method": "sha256 streaming 8 MiB", } atomic_json(paths["intro_receipt"], receipt) print(json.dumps({ "status": receipt["status"], "verified": verified, "duration_seconds": video_duration, "receipt": str(paths["intro_receipt"]), }, ensure_ascii=False)) return 0 if verified else 1 if __name__ == "__main__": try: sys.exit(main()) except Exception as exc: failure = { "status": "failed", "verified": False, "error": f"{type(exc).__name__}: {exc}", "failed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } try: project = json.loads(PROJECT_MANIFEST.read_text(encoding="utf-8")) atomic_json(resolve_active(project, "intro_receipt"), failure) except Exception: pass print(json.dumps(failure, ensure_ascii=False), file=sys.stderr) sys.exit(1)