#!/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] MANIFEST_PATH = ROOT / "script/tts-manifest.json" PROJECT_MANIFEST = ROOT / "script/project-manifest.json" OUTPUT = ROOT / "audio/story-full.wav" RECEIPT = ROOT / "log/tts.json" BASE = "http://192.168.40.33:7862" VOICE = "ngoc-huyen-vbee" 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 atomic_json(path, data): path.parent.mkdir(parents=True, exist_ok=True) temp = path.with_suffix(path.suffix + ".part") temp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") os.replace(temp, path) def get_json(path): with urllib.request.urlopen(BASE + path, timeout=20) as response: if response.status // 100 != 2: raise RuntimeError(f"GET {path} returned 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) streams = info.get("streams", []) audio = [row for row in streams if row.get("codec_type") == "audio"] video = [row for row in 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 streams/duration: {path}") return info, duration def audio_signature(path): with wave.open(str(path), "rb") as source: params = source.getparams() # Frame count varies per segment and is not part of the audio format. return (params.nchannels, params.sampwidth, params.framerate, params.comptype) def synthesize(text, part_path): 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", ) with urllib.request.urlopen(request, timeout=300) as response: if response.status // 100 != 2: raise RuntimeError(f"POST /tts returned HTTP {response.status}") with part_path.open("wb") as handle: while True: block = response.read(1024 * 1024) if not block: break handle.write(block) def main(): plan = json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) project = json.loads(PROJECT_MANIFEST.read_text(encoding="utf-8")) narration = ROOT / project["active_paths"]["spoken_narration"] narration_bytes = narration.read_bytes() if hashlib.sha256(narration_bytes).hexdigest() != plan["input_text_sha256"]: raise RuntimeError("narration hash drift") if "".join(row["text"] for row in plan["segments"]).encode("utf-8") != narration_bytes: raise RuntimeError("segmentation no longer reconstructs narration") health = get_json("/health") voices = get_json("/voices").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: raise RuntimeError("TTS preflight failed") if voice_row.get("model_exists") is not True or voice_row.get("config_exists") is not True: raise RuntimeError("TTS voice model/config unavailable") params = None durations = [] for row in plan["segments"]: output = ROOT / row["output_path"] output.parent.mkdir(parents=True, exist_ok=True) if row.get("status") == "completed" and output.is_file(): info, duration = verify_wav(output) if row.get("artifact_sha256") == sha256(output): durations.append(duration) current = audio_signature(output) params = params or current if current != params: raise RuntimeError("completed segment audio parameters differ") continue part = output.with_suffix(output.suffix + ".part") if part.exists(): try: info, duration = verify_wav(part) os.replace(part, output) except Exception: part.unlink() duration = None else: duration = None if duration is None: last_error = None for attempt in range(1, 4): try: synthesize(row["text"], part) info, duration = verify_wav(part) os.replace(part, output) break except Exception as exc: last_error = exc if part.exists(): part.unlink() if attempt < 3: time.sleep(2 ** attempt) else: row.update({"status": "failed", "error_class": type(last_error).__name__, "error": str(last_error)[:300]}) atomic_json(MANIFEST_PATH, plan) raise last_error current = audio_signature(output) params = params or current if current != params: raise RuntimeError("segment audio parameters differ") row.update({ "status": "completed", "artifact_sha256": sha256(output), "artifact_bytes": output.stat().st_size, "duration_seconds": duration, "sample_rate": current[2], "channels": current[0], }) durations.append(duration) plan["status"] = "rendering" plan["updated_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat() atomic_json(MANIFEST_PATH, plan) temp = OUTPUT.with_suffix(OUTPUT.suffix + ".part") with wave.open(str(temp), "wb") as target: for index, row in enumerate(plan["segments"]): with wave.open(str(ROOT / row["output_path"]), "rb") as source: if index == 0: target.setparams(source.getparams()) elif audio_signature(ROOT / row["output_path"]) != params: raise RuntimeError("concat parameter drift") target.writeframes(source.readframes(source.getnframes())) aggregate_probe, aggregate_duration = verify_wav(temp) os.replace(temp, OUTPUT) aggregate_hash = sha256(OUTPUT) final_probe, final_duration = verify_wav(OUTPUT) if abs(final_duration - sum(durations)) > max(0.1, len(durations) * 0.01): raise RuntimeError("aggregate duration differs from segment sum") now = datetime.datetime.now(datetime.timezone.utc).isoformat() plan.update({"status": "completed", "verified": True, "aggregate_output": str(OUTPUT.relative_to(ROOT)), "aggregate_sha256": aggregate_hash, "aggregate_duration_seconds": final_duration, "updated_at": now}) atomic_json(MANIFEST_PATH, plan) receipt = { "status": "completed", "verified": True, "provider": "piper-wrapper", "base_url": BASE, "endpoint": "/tts", "voice": VOICE, "source_canon_sha256": project["canon_sha256"], "spoken_narration_sha256": plan["input_text_sha256"], "segmentation_plan_sha256": plan["segmentation_plan_sha256"], "segment_count": len(plan["segments"]), "output_path": str(OUTPUT.relative_to(ROOT)), "artifact_sha256": aggregate_hash, "artifact_bytes": OUTPUT.stat().st_size, "duration_seconds": final_duration, "segment_duration_sum": sum(durations), "sample_rate": int(final_probe["streams"][0]["sample_rate"]), "channels": final_probe["streams"][0]["channels"], "probe": final_probe, "health_checked_at": now, "created_at": now, } atomic_json(RECEIPT, receipt) project["steps"]["tts"] = "completed" project["tts"] = {"status": "completed", "verified": True, "receipt": str(RECEIPT.relative_to(ROOT)), "artifact_sha256": aggregate_hash, "duration_seconds": final_duration, "segment_count": len(plan["segments"]), "completed_at": now} project["story_duration_seconds"] = final_duration project["updated_at"] = now atomic_json(PROJECT_MANIFEST, project) print(json.dumps({"status": "completed", "segments": len(plan["segments"]), "duration_seconds": final_duration, "sha256": aggregate_hash}, ensure_ascii=False)) if __name__ == "__main__": main()