#!/usr/bin/env python3 from __future__ import annotations import datetime import hashlib import json import os import subprocess import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] PROJECT_MANIFEST = ROOT / "script/project-manifest.json" TEXT = "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_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 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")) recovery = project.get("intro_encoder_recovery") or {} paths = {key: resolve_active(project, key) for key in ( "promotion_receipt", "layout_receipt", "intro_normalized", "intro_voice", "intro_silence", "intro_audio", "intro", "intro_receipt", )} old_receipt = ROOT / recovery.get("from_receipt", "") if recovery.get("failed_encoder") != "h264_nvenc" or recovery.get("recovery_encoder") != "libx264" or recovery.get("tts_reused") is not True: raise RuntimeError("intro encoder recovery contract is absent") failure = json.loads(old_receipt.read_text(encoding="utf-8")) if failure.get("status") != "failed" or "Cannot load libcuda" not in failure.get("error", "") and "h264_nvenc" not in failure.get("error", ""): raise RuntimeError("preserved receipt does not authorize encoder-only recovery") 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("verified") is not True or layout.get("verified") is not True or layout.get("source_canon_sha256") != canon_hash: raise RuntimeError("promotion/layout gate is absent or stale") required = [paths["intro_normalized"], paths["intro_voice"], paths["intro_silence"], paths["intro_audio"]] missing = [str(path) for path in required if not path.exists() or path.stat().st_size <= 0] if missing: raise RuntimeError(f"verified intro recovery inputs missing: {missing}") if paths["intro"].exists() or paths["intro_receipt"].exists(): raise RuntimeError("intro v2 output/receipt already exists and requires audit") voice_probe = probe(paths["intro_voice"]) silence_probe = probe(paths["intro_silence"]) audio_probe = probe(paths["intro_audio"]) voice_duration = float(voice_probe["format"]["duration"]) silence_duration = float(silence_probe["format"]["duration"]) audio_duration = float(audio_probe["format"]["duration"]) if abs(silence_duration - 2.0) > 0.001 or abs(audio_duration - voice_duration - 2.0) > 0.02: raise RuntimeError("intro recovery audio duration verification failed") if not all(len([item for item in info.get("streams", []) if item.get("codec_type") == "audio"]) == 1 for info in (voice_probe, silence_probe, audio_probe)): raise RuntimeError("intro recovery audio stream verification failed") paths["intro"].parent.mkdir(parents=True, exist_ok=True) 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", "libx264", "-preset", "medium", "-crf", "18", "-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", []) videos = [item for item in streams if item.get("codec_type") == "video"] audios = [item for item in streams if item.get("codec_type") == "audio"] verified = ( abs(video_duration - audio_duration) <= 1 / 30 + 0.005 and len(videos) == 1 and len(audios) == 1 and videos[0].get("codec_name") == "h264" and (int(videos[0].get("width", 0)), int(videos[0].get("height", 0))) == (1920, 1080) and audios[0].get("codec_name") == "aac" ) receipt = { "status": "completed" if verified else "failed", "verified": verified, "source_canon_sha256": canon_hash, "recovery_mode": "encode_only", "preserved_failure_receipt": {"path": str(old_receipt), "sha256": sha256(old_receipt)}, "tts_reused_without_new_request": True, "intro_text": TEXT, "input_path": str(paths["intro_normalized"]), "input_sha256": sha256(paths["intro_normalized"]), "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, "encoder": "libx264", "output_path": str(paths["intro"]), "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: print(json.dumps({"status": "failed", "verified": False, "error": f"{type(exc).__name__}: {exc}"}, ensure_ascii=False), file=sys.stderr) sys.exit(1)