#!/usr/bin/env python3 import datetime import hashlib import json import os import subprocess import time import urllib.error 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" CANONICAL_INTRO = "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 sha_bytes(data): return hashlib.sha256(data).hexdigest() def sha_path(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 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(path): with urllib.request.urlopen(BASE + path, timeout=30) as response: if response.status // 100 != 2: raise RuntimeError(f"GET {path} HTTP {response.status}") return json.load(response) def probe(path): result = subprocess.run( [ "ffprobe", "-v", "error", "-show_entries", "format=duration,size", "-show_entries", "stream=index,codec_type,codec_name,sample_rate,channels,sample_fmt", "-of", "json", str(path), ], check=True, capture_output=True, text=True, ) return json.loads(result.stdout) def verify_wav(path): raw = path.open("rb").read(12) if len(raw) != 12 or raw[:4] != b"RIFF" or raw[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 WAV stream/duration: {path}") return info, duration def request_body(text): return urllib.parse.urlencode({"text": text, "voice": VOICE}).encode("utf-8") def synthesize(text, part): request = urllib.request.Request( BASE + "/tts", data=request_body(text), headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST", ) with urllib.request.urlopen(request, timeout=360) as response, part.open("wb") as handle: if response.status // 100 != 2: raise RuntimeError(f"POST /tts HTTP {response.status}") content_type = response.headers.get_content_type() if content_type not in {"audio/wav", "audio/x-wav", "audio/wave", "application/octet-stream"}: raise RuntimeError(f"unexpected TTS content type: {content_type}") for block in iter(lambda: response.read(1024 * 1024), b""): handle.write(block) def main(): manifest_path = ROOT / "script/project-manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) paths = manifest["active_paths"] pronunciation_path = ROOT / paths["tts_pronunciation_receipt"] narration_receipt_path = ROOT / paths["tts_receipt"] voice_path = ROOT / paths["intro_voice"] silence_path = ROOT / paths["intro_silence"] full_path = ROOT / paths["intro_audio"] receipt_path = ROOT / paths["intro_audio_receipt"] if manifest.get("steps", {}).get("tts") != "completed": raise RuntimeError("narration TTS must be terminal before intro voice") for output in (voice_path, silence_path, full_path, receipt_path): if output.exists() or output.with_suffix(output.suffix + ".part").exists(): raise RuntimeError(f"intro audio output must be virgin: {output}") pronunciation = json.loads(pronunciation_path.read_text(encoding="utf-8")) narration_receipt = json.loads(narration_receipt_path.read_text(encoding="utf-8")) if pronunciation.get("status") != "completed" or pronunciation.get("verified") is not True: raise RuntimeError("pronunciation receipt not terminal") if narration_receipt.get("status") != "completed" or narration_receipt.get("verified") is not True: raise RuntimeError("narration TTS receipt not terminal") if narration_receipt.get("tts_pronunciation_receipt_sha256") != sha_path(pronunciation_path): raise RuntimeError("intro pronunciation lineage drift") intro = pronunciation.get("intro", {}) source_text = intro.get("source_text") tts_text = intro.get("tts_text") if source_text != CANONICAL_INTRO: raise RuntimeError("canonical intro source drift") if intro.get("source_text_sha256") != sha_bytes(source_text.encode("utf-8")): raise RuntimeError("canonical intro source hash drift") if not tts_text or intro.get("tts_text_sha256") != sha_bytes(tts_text.encode("utf-8")): raise RuntimeError("intro TTS text/hash invalid") if intro.get("replacement_count", 0) < 3 or intro.get("reverse_reconstruction_verified") is not True: raise RuntimeError("intro pronunciation transform not verified") health = get_json("/health") voices_payload = get_json("/voices") voices = voices_payload.get("voices", voices_payload if isinstance(voices_payload, list) else []) voice_row = next((row for row in voices if row.get("id") == VOICE), None) if health.get("status") != "ok" or not voice_row: raise RuntimeError("TTS preflight failed") if voice_row.get("model_exists") is False or voice_row.get("config_exists") is False: raise RuntimeError("TTS voice model/config unavailable") voice_path.parent.mkdir(parents=True, exist_ok=True) part = voice_path.with_suffix(voice_path.suffix + ".part") attempts = [] error = None for attempt in range(1, 4): try: synthesize(tts_text, part) verify_wav(part) os.replace(part, voice_path) attempts.append({"attempt": attempt, "status": "completed"}) break except urllib.error.HTTPError as exc: attempts.append({"attempt": attempt, "status": "failed", "error_class": type(exc).__name__, "http_status": exc.code}) error = exc part.unlink(missing_ok=True) if 400 <= exc.code < 500: break if attempt < 3: time.sleep(2 ** attempt) except Exception as exc: attempts.append({"attempt": attempt, "status": "failed", "error_class": type(exc).__name__}) error = exc part.unlink(missing_ok=True) if attempt < 3: time.sleep(2 ** attempt) if not voice_path.exists(): raise error voice_probe, voice_duration = verify_wav(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_part = silence_path.with_suffix(silence_path.suffix + ".part") silence_frames = params.framerate * 2 with wave.open(str(silence_part), "wb") as silence: silence.setparams((params.nchannels, params.sampwidth, params.framerate, 0, params.comptype, params.compname)) silence.writeframes(b"\x00" * silence_frames * params.nchannels * params.sampwidth) verify_wav(silence_part) os.replace(silence_part, silence_path) full_part = full_path.with_suffix(full_path.suffix + ".part") with wave.open(str(full_part), "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) verify_wav(full_part) os.replace(full_part, full_path) silence_probe, silence_duration = verify_wav(silence_path) full_probe, full_duration = verify_wav(full_path) if abs(silence_duration - 2.0) > 0.0001 or abs(full_duration - voice_duration - 2.0) > 0.001: raise RuntimeError("intro duration relation failed") body = request_body(tts_text) 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"], "request_content_type": "application/x-www-form-urlencoded", "payload_sha256": sha_bytes(body), "attempts": attempts, "pronunciation_receipt_path": paths["tts_pronunciation_receipt"], "pronunciation_receipt_sha256": sha_path(pronunciation_path), "source_canon_sha256": pronunciation["source_canon_sha256"], "intro_source_text": source_text, "intro_source_text_sha256": intro["source_text_sha256"], "intro_tts_text": tts_text, "intro_tts_text_sha256": intro["tts_text_sha256"], "voice_path": paths["intro_voice"], "voice_sha256": sha_path(voice_path), "voice_duration_seconds": voice_duration, "silence_path": paths["intro_silence"], "silence_sha256": sha_path(silence_path), "silence_duration_seconds": silence_duration, "intro_full_path": paths["intro_audio"], "intro_full_sha256": sha_path(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_path, receipt) manifest["intro_audio"] = { "status": "completed", "verified": True, "receipt": paths["intro_audio_receipt"], "artifact_sha256": receipt["intro_full_sha256"], "duration_seconds": full_duration, "completed_at": now, } manifest["updated_at"] = now atomic_json(manifest_path, manifest) 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()