#!/usr/bin/env python3 import datetime import hashlib import json import os import subprocess import time import urllib.parse import urllib.request import wave from pathlib import Path ROOT = Path(__file__).resolve().parents[1] BASE = "http://192.168.40.33:7862" VOICE = "ngoc-huyen-vbee" 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." VOICE_PATH = ROOT / "audio/intro-voice.wav" SILENCE_PATH = ROOT / "audio/intro-silence-2s.wav" FULL_PATH = ROOT / "audio/intro-full.wav" RECEIPT = ROOT / "log/intro-audio.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,sample_rate,channels,sample_fmt", "-of", "json", str(path)], capture_output=True, text=True, check=True, ) return json.loads(result.stdout) def verify(path): if path.open("rb").read(12)[:4] != b"RIFF" or path.open("rb").read(12)[8:12] != b"WAVE": raise RuntimeError(f"invalid RIFF/WAVE: {path}") info = probe(path) audio = [row for row in info.get("streams", []) if row.get("codec_type") == "audio"] video = [row for row in info.get("streams", []) if row.get("codec_type") == "video"] duration = float(info.get("format", {}).get("duration", 0)) if len(audio) != 1 or video or duration <= 0: raise RuntimeError(f"invalid audio stream/duration: {path}") return info, duration 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(): project = json.loads((ROOT / "script/project-manifest.json").read_text(encoding="utf-8")) if project.get("steps", {}).get("tts") != "completed": raise RuntimeError("narration TTS must be terminal before intro voice") if any(path.exists() for path in (VOICE_PATH, SILENCE_PATH, FULL_PATH, RECEIPT)): raise RuntimeError("intro audio outputs must be virgin") with urllib.request.urlopen(BASE + "/health", timeout=20) as response: health = json.load(response) with urllib.request.urlopen(BASE + "/voices", timeout=20) as response: voices = json.load(response).get("voices", []) voice_row = next((row for row in voices if row.get("id") == VOICE), None) if health.get("status") != "ok" or not voice_row or voice_row.get("model_exists") is not True or voice_row.get("config_exists") is not True: raise RuntimeError("TTS preflight failed") part = VOICE_PATH.with_suffix(".wav.part") request = urllib.request.Request( BASE + "/tts", data=urllib.parse.urlencode({"text": TEXT, "voice": VOICE}).encode("utf-8"), headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST", ) last_error = None for attempt in range(1, 4): try: with urllib.request.urlopen(request, timeout=300) as response, part.open("wb") as handle: while True: block = response.read(1024 * 1024) if not block: break handle.write(block) verify(part) os.replace(part, VOICE_PATH) break except Exception as exc: last_error = exc if part.exists(): part.unlink() if attempt < 3: time.sleep(2 ** attempt) else: raise last_error voice_probe, voice_duration = verify(VOICE_PATH) with wave.open(str(VOICE_PATH), "rb") as source: params = source.getparams() if params.comptype != "NONE": raise RuntimeError("intro voice is not uncompressed PCM") voice_frames = source.readframes(source.getnframes()) silence_frames = params.framerate * 2 with wave.open(str(SILENCE_PATH), "wb") as silence: silence.setparams((params.nchannels, params.sampwidth, params.framerate, silence_frames, params.comptype, params.compname)) silence.writeframes(b"\x00" * silence_frames * params.nchannels * params.sampwidth) with wave.open(str(FULL_PATH), "wb") as full: full.setparams((params.nchannels, params.sampwidth, params.framerate, 0, params.comptype, params.compname)) full.writeframes(voice_frames) full.writeframes(b"\x00" * silence_frames * params.nchannels * params.sampwidth) silence_probe, silence_duration = verify(SILENCE_PATH) full_probe, full_duration = verify(FULL_PATH) if abs(silence_duration - 2.0) > 0.0001 or abs(full_duration - voice_duration - 2.0) > 0.001: raise RuntimeError("intro silence/full duration relation failed") now = datetime.datetime.now(datetime.timezone.utc).isoformat() receipt = { "status": "completed", "verified": True, "provider": "piper-wrapper", "base_url": BASE, "endpoint": "/tts", "voice": VOICE, "request_fields": ["text", "voice"], "intro_text": TEXT, "intro_text_sha256": hashlib.sha256(TEXT.encode("utf-8")).hexdigest(), "voice_path": str(VOICE_PATH.relative_to(ROOT)), "voice_sha256": sha256(VOICE_PATH), "voice_duration_seconds": voice_duration, "silence_path": str(SILENCE_PATH.relative_to(ROOT)), "silence_sha256": sha256(SILENCE_PATH), "silence_duration_seconds": silence_duration, "intro_full_path": str(FULL_PATH.relative_to(ROOT)), "intro_full_sha256": sha256(FULL_PATH), "intro_full_duration_seconds": full_duration, "voice_probe": voice_probe, "silence_probe": silence_probe, "intro_full_probe": full_probe, "completed_at": now, } atomic_json(RECEIPT, receipt) project["active_paths"].update({"intro_voice": str(VOICE_PATH.relative_to(ROOT)), "intro_silence": str(SILENCE_PATH.relative_to(ROOT)), "intro_audio": str(FULL_PATH.relative_to(ROOT)), "intro_audio_receipt": str(RECEIPT.relative_to(ROOT))}) project["intro_audio"] = {"status": "completed", "verified": True, "receipt": str(RECEIPT.relative_to(ROOT)), "artifact_sha256": receipt["intro_full_sha256"], "duration_seconds": full_duration, "completed_at": now} project["updated_at"] = now atomic_json(ROOT / "script/project-manifest.json", project) print(json.dumps({"status": "completed", "voice_duration": voice_duration, "full_duration": full_duration, "sha256": receipt["intro_full_sha256"]}, ensure_ascii=False)) if __name__ == "__main__": main()