#!/usr/bin/env python3 import datetime import hashlib import json import os import subprocess import time import urllib.parse import urllib.error 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" 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, 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) streams = info.get("streams", []) audio = [x for x in streams if x.get("codec_type") == "audio"] video = [x for x in streams if x.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 signature(path): with wave.open(str(path), "rb") as source: params = source.getparams() return params.nchannels, params.sampwidth, params.framerate, params.comptype def request_body(text): return urllib.parse.urlencode({"text": text, "voice": VOICE}).encode("utf-8") def synthesize(text, part): body = request_body(text) request = urllib.request.Request(BASE + "/tts", data=body, 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}") while True: block = response.read(1024 * 1024) if not block: break 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"] plan_path = ROOT / paths["tts_manifest"] plan = json.loads(plan_path.read_text(encoding="utf-8")) narration = ROOT / paths["spoken_narration"] source = ROOT / paths["spoken_narration_source"] lexicon = ROOT / paths["tts_pronunciation_lexicon"] pronunciation_path = ROOT / paths["tts_pronunciation_receipt"] receipt_path = ROOT / paths["tts_receipt"] output = ROOT / paths["tts_audio"] pronunciation = json.loads(pronunciation_path.read_text(encoding="utf-8")) narration_bytes = narration.read_bytes() if hashlib.sha256(narration_bytes).hexdigest() != plan["input_text_sha256"]: raise RuntimeError("narration hash drift") if "".join(x["text"] for x in plan["segments"]).encode("utf-8") != narration_bytes: raise RuntimeError("segmentation does not reconstruct normalized narration") checks = { "source_canon_sha256": pronunciation["source_canon_sha256"], "spoken_narration_source_sha256": sha256(source), "lexicon_sha256": sha256(lexicon), "tts_pronunciation_receipt_sha256": sha256(pronunciation_path), } for key, value in checks.items(): if plan.get(key) != value: raise RuntimeError(f"TTS plan lineage drift: {key}") if pronunciation.get("verified") is not True or pronunciation.get("spoken_narration_sha256") != plan["input_text_sha256"]: raise RuntimeError("pronunciation gate not terminal on TTS input") if receipt_path.exists() or output.exists(): raise RuntimeError("aggregate output/receipt must be virgin") 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((x for x in voices if x.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") fmt = None durations = [] for row in plan["segments"]: attempts = list(row.get("attempts", [])) expected_payload_sha = hashlib.sha256(request_body(row["text"])).hexdigest() if row.get("text_sha256") != hashlib.sha256(row["text"].encode("utf-8")).hexdigest(): raise RuntimeError("segment text hash drift") if row.get("payload_sha256") != expected_payload_sha: raise RuntimeError("segment payload hash drift") if row.get("endpoint") != BASE + "/tts" or row.get("voice") != VOICE or row.get("content_type") != "application/x-www-form-urlencoded": raise RuntimeError("segment request contract drift") segment = ROOT / row["output_path"] segment.parent.mkdir(parents=True, exist_ok=True) if row.get("status") == "completed" and segment.is_file() and row.get("artifact_sha256") == sha256(segment) and row.get("terminal_payload_sha256") == expected_payload_sha: _, duration = verify_wav(segment) current = signature(segment) fmt = fmt or current if current != fmt: raise RuntimeError("completed segment format drift") durations.append(duration) continue part = segment.with_suffix(segment.suffix + ".part") duration = None if part.exists(): try: _, duration = verify_wav(part) os.replace(part, segment) attempts.append({"attempt": 0, "status": "recovered_part"}) except Exception: part.unlink(missing_ok=True) duration = None if duration is None: error = None for attempt in range(1, 4): try: synthesize(row["text"], part) _, duration = verify_wav(part) os.replace(part, segment) 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 duration is None: row.update({"status": "failed", "terminal_payload_sha256": expected_payload_sha, "attempts": attempts, "error_class": type(error).__name__, "error": str(error)[:300]}) plan["status"] = "failed" plan["updated_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat() atomic_json(plan_path, plan) raise error current = signature(segment) fmt = fmt or current if current != fmt: raise RuntimeError("segment audio format differs") row.update({"status": "completed", "terminal_payload_sha256": expected_payload_sha, "attempts": attempts, "artifact_sha256": sha256(segment), "artifact_bytes": segment.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(plan_path, plan) temp = output.with_suffix(output.suffix + ".part") with wave.open(str(temp), "wb") as target: for index, row in enumerate(plan["segments"]): segment = ROOT / row["output_path"] with wave.open(str(segment), "rb") as source_wav: if index == 0: target.setparams(source_wav.getparams()) elif signature(segment) != fmt: raise RuntimeError("concat format drift") target.writeframes(source_wav.readframes(source_wav.getnframes())) aggregate_probe, aggregate_duration = verify_wav(temp) if abs(aggregate_duration - sum(durations)) > max(0.1, len(durations) * 0.01): temp.unlink(missing_ok=True) raise RuntimeError("aggregate duration mismatch") os.replace(temp, output) now = datetime.datetime.now(datetime.timezone.utc).isoformat() artifact_sha = sha256(output) plan.update({"status": "completed", "verified": True, "aggregate_output": paths["tts_audio"], "aggregate_sha256": artifact_sha, "aggregate_duration_seconds": aggregate_duration, "updated_at": now}) atomic_json(plan_path, plan) 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", "source_canon_sha256": plan["source_canon_sha256"], "spoken_narration_source_sha256": plan["spoken_narration_source_sha256"], "spoken_narration_sha256": plan["input_text_sha256"], "tts_pronunciation_receipt_path": paths["tts_pronunciation_receipt"], "tts_pronunciation_receipt_sha256": plan["tts_pronunciation_receipt_sha256"], "lexicon_sha256": plan["lexicon_sha256"], "segmentation_plan_sha256": plan["segmentation_plan_sha256"], "segment_count": len(plan["segments"]), "output_path": paths["tts_audio"], "artifact_sha256": artifact_sha, "artifact_bytes": output.stat().st_size, "duration_seconds": aggregate_duration, "segment_duration_sum": sum(durations), "sample_rate": int(aggregate_probe["streams"][0]["sample_rate"]), "channels": aggregate_probe["streams"][0]["channels"], "probe": aggregate_probe, "health_checked_at": now, "created_at": now, } atomic_json(receipt_path, receipt) manifest["steps"]["tts"] = "completed" manifest["tts"] = {"status": "completed", "verified": True, "receipt": paths["tts_receipt"], "artifact_sha256": artifact_sha, "duration_seconds": aggregate_duration, "segment_count": len(plan["segments"]), "completed_at": now} manifest["story_duration_seconds"] = aggregate_duration manifest["updated_at"] = now atomic_json(manifest_path, manifest) print(json.dumps({"status": "completed", "segments": len(plan["segments"]), "duration_seconds": aggregate_duration, "sha256": artifact_sha}, ensure_ascii=False)) if __name__ == "__main__": main()