#!/usr/bin/env python3 import hashlib import json import os import subprocess import sys import time import urllib.parse import urllib.request import wave from datetime import datetime, timezone from pathlib import Path PROJECT = Path(__file__).resolve().parents[1] CANON = PROJECT / "story/story-canon.txt" SPOKEN = PROJECT / "story/spoken-narration.txt" SOURCE = PROJECT / "story/tts-pronunciation.txt" PRONUNCIATION_RECEIPT = PROJECT / "log/tts-pronunciation.json" LEXICON = PROJECT / "script/pronunciation-lexicon.json" SEGMENT_DIR = PROJECT / "audio/pronunciation-segments" OUTPUT = PROJECT / "audio/story-full.wav" MANIFEST = PROJECT / "script/tts-manifest.json" RECEIPT = PROJECT / "log/tts-verification.json" BASE = "http://192.168.40.33:7862" VOICE = "ngoc-huyen-vbee" MAX_CHARS = 1800 def sha_bytes(data): return hashlib.sha256(data).hexdigest() def sha_file(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 chunks_exact(text): chunks = [] start = 0 while start < len(text): end = min(start + MAX_CHARS, len(text)) if end < len(text): candidates = [text.rfind(token, start, end) for token in ("\n\n", ". ", "? ", "! ", "; ", ", ", " ")] cut = max(candidates) if cut > start + MAX_CHARS // 2: token_len = 2 if text[cut:cut + 2] in {"\n\n", ". ", "? ", "! ", "; ", ", "} else 1 end = cut + token_len chunks.append(text[start:end]) start = end if "".join(chunks) != text: raise RuntimeError("TTS chunk reconstruction mismatch") return chunks def probe(path): result = subprocess.run([ "ffprobe", "-v", "error", "-show_entries", "format=duration,size", "-show_entries", "stream=index,codec_name,sample_rate,channels", "-of", "json", str(path), ], capture_output=True, text=True, check=True) return json.loads(result.stdout) def wav_info(path): with wave.open(str(path), "rb") as handle: return { "channels": handle.getnchannels(), "sample_width": handle.getsampwidth(), "sample_rate": handle.getframerate(), "frames": handle.getnframes(), "compression": handle.getcomptype(), "duration_seconds": handle.getnframes() / handle.getframerate(), } def preflight(): with urllib.request.urlopen(BASE + "/health", timeout=20) as response: health = json.load(response) if health.get("status") != "ok": raise RuntimeError("Piper health check failed") with urllib.request.urlopen(BASE + "/voices", timeout=20) as response: voices = json.load(response).get("voices", []) row = next((item for item in voices if item.get("id") == VOICE), None) if not row or row.get("model_exists") is not True or row.get("config_exists") is not True: raise RuntimeError("ngoc-huyen-vbee model/config unavailable") return health, row def synthesize(text, temp): body = urllib.parse.urlencode({"text": text, "voice": VOICE}).encode("utf-8") request = urllib.request.Request( BASE + "/tts", data=body, headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST", ) last = None for attempt in range(1, 4): try: with urllib.request.urlopen(request, timeout=300) as response: data = response.read() if len(data) < 44 or data[:4] != b"RIFF" or data[8:12] != b"WAVE": raise RuntimeError("Piper response is not WAV") temp.write_bytes(data) info = wav_info(temp) if info["duration_seconds"] <= 0 or info["compression"] != "NONE": raise RuntimeError("WAV verification failed") return attempt, info except Exception as exc: last = exc if temp.exists(): temp.unlink() if attempt < 3: time.sleep(2 ** attempt) raise RuntimeError(f"TTS request failed after retries: {last}") def concatenate(paths, output): first = wav_info(paths[0]) params = (first["channels"], first["sample_width"], first["sample_rate"]) temp = output.with_suffix(".part.wav") with wave.open(str(temp), "wb") as target: target.setnchannels(params[0]) target.setsampwidth(params[1]) target.setframerate(params[2]) for path in paths: with wave.open(str(path), "rb") as source: current = (source.getnchannels(), source.getsampwidth(), source.getframerate()) if current != params or source.getcomptype() != "NONE": raise RuntimeError(f"Segment WAV format mismatch: {path}") while True: frames = source.readframes(65536) if not frames: break target.writeframesraw(frames) os.replace(temp, output) def main(): required = (CANON, SPOKEN, SOURCE, PRONUNCIATION_RECEIPT, LEXICON) if any(not path.is_file() for path in required): raise RuntimeError("Pronunciation-ready TTS prerequisites missing") pronunciation = json.loads(PRONUNCIATION_RECEIPT.read_text(encoding="utf-8")) if pronunciation.get("verified") is not True or pronunciation.get("semantic_content_changed") is not False: raise RuntimeError("Pronunciation receipt is not verified") canon_sha = sha_file(CANON) spoken_sha = sha_file(SPOKEN) pronunciation_sha = sha_file(SOURCE) if pronunciation.get("source_canon_sha256") != canon_sha: raise RuntimeError("Pronunciation receipt canon hash mismatch") if pronunciation.get("source_sha256") != spoken_sha: raise RuntimeError("Pronunciation receipt spoken hash mismatch") if pronunciation.get("projection_sha256") != pronunciation_sha: raise RuntimeError("Pronunciation receipt projection hash mismatch") if pronunciation.get("lexicon_sha256") != sha_file(LEXICON): raise RuntimeError("Pronunciation receipt lexicon hash mismatch") text = SOURCE.read_text(encoding="utf-8") source_sha = pronunciation_sha health, voice_row = preflight() chunks = chunks_exact(text) SEGMENT_DIR.mkdir(parents=True, exist_ok=True) existing = {} if MANIFEST.exists(): old = json.loads(MANIFEST.read_text(encoding="utf-8")) if old.get("source_sha256") == source_sha and old.get("voice") == VOICE: existing = {row["index"]: row for row in old.get("segments", [])} rows = [] paths = [] for index, chunk in enumerate(chunks, 1): output = SEGMENT_DIR / f"segment-{index:03d}.wav" text_sha = sha_bytes(chunk.encode("utf-8")) prior = existing.get(index, {}) if output.exists() and prior.get("verified") is True and prior.get("text_sha256") == text_sha and prior.get("output_sha256") == sha_file(output): info = wav_info(output) attempts = prior.get("attempts", 1) else: if output.exists(): raise RuntimeError(f"Stale unversioned TTS segment exists: {output}") temp = output.with_suffix(".part.wav") attempts, info = synthesize(chunk, temp) os.replace(temp, output) row = { "index": index, "characters": len(chunk), "words": len(chunk.split()), "text_sha256": text_sha, "output": str(output.relative_to(PROJECT)), "output_sha256": sha_file(output), "bytes": output.stat().st_size, "duration_seconds": info["duration_seconds"], "format": info, "attempts": attempts, "verified": True, } rows.append(row) paths.append(output) MANIFEST.write_text(json.dumps({ "version": 1, "status": "running", "source": str(SOURCE.relative_to(PROJECT)), "source_sha256": source_sha, "voice": VOICE, "base_url": BASE, "reconstruction_verified": "".join(chunks) == text, "segments_total": len(chunks), "segments_completed": len(rows), "segments": rows, }, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"segment": index, "total": len(chunks), "duration": round(info["duration_seconds"], 3)}, ensure_ascii=False), flush=True) concatenate(paths, OUTPUT) output_info = wav_info(OUTPUT) output_probe = probe(OUTPUT) expected_duration = sum(row["duration_seconds"] for row in rows) verified = ( "".join(chunks) == text and len(rows) == len(chunks) and abs(output_info["duration_seconds"] - expected_duration) < 0.05 and len(output_probe.get("streams", [])) == 1 and output_probe["streams"][0].get("codec_name") == "pcm_s16le" ) manifest = { "version": 1, "status": "completed" if verified else "failed", "source": str(SOURCE.relative_to(PROJECT)), "source_sha256": source_sha, "source_canon_sha256": canon_sha, "spoken_narration_sha256": spoken_sha, "pronunciation_receipt": str(PRONUNCIATION_RECEIPT.relative_to(PROJECT)), "pronunciation_receipt_sha256": sha_file(PRONUNCIATION_RECEIPT), "pronunciation_lexicon_sha256": sha_file(LEXICON), "voice": VOICE, "base_url": BASE, "reconstruction_verified": True, "segments_total": len(rows), "segments_completed": len(rows), "segments": rows, "output": str(OUTPUT.relative_to(PROJECT)), "output_sha256": sha_file(OUTPUT), "duration_seconds": output_info["duration_seconds"], "completed_at": datetime.now(timezone.utc).isoformat(), } MANIFEST.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") receipt = { "version": 1, "verified": verified, "status": manifest["status"], "service": health.get("service"), "base_url": BASE, "voice": VOICE, "voice_verification": voice_row, "source_path": str(SOURCE.relative_to(PROJECT)), "source_sha256": source_sha, "source_canon_sha256": canon_sha, "spoken_narration_path": str(SPOKEN.relative_to(PROJECT)), "spoken_narration_sha256": spoken_sha, "pronunciation_receipt": str(PRONUNCIATION_RECEIPT.relative_to(PROJECT)), "pronunciation_receipt_sha256": sha_file(PRONUNCIATION_RECEIPT), "pronunciation_lexicon_sha256": sha_file(LEXICON), "pronunciation_verified": True, "chunk_reconstruction_sha256": sha_bytes("".join(chunks).encode("utf-8")), "segments_total": len(rows), "segments_completed": len(rows), "output": str(OUTPUT.relative_to(PROJECT)), "output_sha256": manifest["output_sha256"], "duration_seconds": output_info["duration_seconds"], "probe": output_probe, "checked_at": datetime.now(timezone.utc).isoformat(), } RECEIPT.parent.mkdir(parents=True, exist_ok=True) RECEIPT.write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"verified": verified, "segments": len(rows), "duration_seconds": output_info["duration_seconds"], "output_sha256": receipt["output_sha256"]}, ensure_ascii=False)) return 0 if verified else 1 if __name__ == "__main__": try: sys.exit(main()) except Exception as exc: print(f"TTS blocked: {exc}", file=sys.stderr) sys.exit(1)