#!/usr/bin/env python3 import datetime import hashlib import json import os import re import urllib.parse from pathlib import Path ROOT = Path(__file__).resolve().parents[1] BASE = "http://192.168.40.33:7862" VOICE = "ngoc-huyen-vbee" TARGET_CHARS = 2200 MAX_CHARS = 3000 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 split_text(text): paragraphs = re.findall(r".*?(?:\n\n|\Z)", text, flags=re.DOTALL) paragraphs = [p for p in paragraphs if p] chunks = [] current = "" for paragraph in paragraphs: if len(paragraph) > MAX_CHARS: sentences = re.findall(r".*?(?:[.!?…][”’']?\s+|\Z)", paragraph, flags=re.DOTALL) units = [s for s in sentences if s] else: units = [paragraph] for unit in units: if current and len(current) + len(unit) > TARGET_CHARS: chunks.append(current) current = "" if len(unit) > MAX_CHARS: raise RuntimeError("sentence exceeds segmentation maximum") current += unit if current: chunks.append(current) if "".join(chunks) != text: raise RuntimeError("segmentation reconstruction failed") if not chunks or any(not x or len(x) > MAX_CHARS for x in chunks): raise RuntimeError("invalid segmentation plan") return chunks def main(): manifest_path = ROOT / "script/project-manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) paths = manifest["active_paths"] output = ROOT / "script/tts-manifest.json" if output.exists() or any((ROOT / "audio/segments").glob("segment-*.wav*")): raise RuntimeError("TTS plan/segments must be virgin") if manifest.get("steps", {}).get("pronunciation") != "completed": raise RuntimeError("pronunciation must be terminal") promotion_path = ROOT / paths["promotion_receipt"] pronunciation_path = ROOT / paths["tts_pronunciation_receipt"] promotion = json.loads(promotion_path.read_text(encoding="utf-8")) pronunciation = json.loads(pronunciation_path.read_text(encoding="utf-8")) if promotion.get("verified") is not True or pronunciation.get("verified") is not True: raise RuntimeError("upstream gate not terminal") narration_path = ROOT / paths["spoken_narration"] source_path = ROOT / paths["spoken_narration_source"] lexicon_path = ROOT / paths["tts_pronunciation_lexicon"] narration_bytes = narration_path.read_bytes() narration_sha = sha_bytes(narration_bytes) if narration_sha != pronunciation.get("spoken_narration_sha256"): raise RuntimeError("normalized narration hash drift") if sha_path(source_path) != pronunciation.get("spoken_narration_source_sha256"): raise RuntimeError("narration source hash drift") if sha_path(lexicon_path) != pronunciation.get("lexicon_sha256"): raise RuntimeError("lexicon hash drift") if promotion.get("source_canon_sha256") != pronunciation.get("source_canon_sha256"): raise RuntimeError("canon lineage drift") chunks = split_text(narration_bytes.decode("utf-8")) boundaries = [] cursor = 0 for chunk in chunks: boundaries.append((cursor, cursor + len(chunk))) cursor += len(chunk) split_replacements = [] for row in pronunciation.get("replacements", []): start, end = row["output_char_span"] if not any(left <= start and end <= right for left, right in boundaries): split_replacements.append(row) if split_replacements: raise RuntimeError("segmentation cuts a pronunciation replacement span") segments = [] for index, text in enumerate(chunks, 1): raw = text.encode("utf-8") payload = urllib.parse.urlencode({"text": text, "voice": VOICE}).encode("utf-8") segments.append({ "index": index, "text": text, "text_sha256": sha_bytes(raw), "payload_sha256": sha_bytes(payload), "endpoint": BASE + "/tts", "voice": VOICE, "content_type": "application/x-www-form-urlencoded", "byte_count": len(raw), "output_path": f"audio/segments/segment-{index:04d}.wav", "status": "pending", }) basis = { "input_text_sha256": narration_sha, "endpoint": BASE + "/tts", "voice": VOICE, "segments": [{"index": x["index"], "text_sha256": x["text_sha256"], "payload_sha256": x["payload_sha256"], "endpoint": x["endpoint"], "voice": x["voice"], "content_type": x["content_type"], "byte_count": x["byte_count"], "output_path": x["output_path"]} for x in segments], } now = datetime.datetime.now(datetime.timezone.utc).isoformat() plan = { "status": "locked", "verified": True, "input_path": paths["spoken_narration"], "input_text_sha256": narration_sha, "source_canon_sha256": pronunciation["source_canon_sha256"], "spoken_narration_source_sha256": pronunciation["spoken_narration_source_sha256"], "lexicon_sha256": pronunciation["lexicon_sha256"], "tts_pronunciation_receipt_path": paths["tts_pronunciation_receipt"], "tts_pronunciation_receipt_sha256": sha_path(pronunciation_path), "provider": "piper-wrapper", "base_url": BASE, "endpoint": "/tts", "voice": VOICE, "request_fields": ["text", "voice"], "request_content_type": "application/x-www-form-urlencoded", "pronunciation_spans_preserved": True, "segment_count": len(segments), "segmentation_plan_sha256": sha_bytes(json.dumps(basis, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")), "segments": segments, "created_at": now, "updated_at": now, } atomic_json(output, plan) manifest["active_paths"]["tts_manifest"] = "script/tts-manifest.json" manifest["tts_plan"] = {"status": "locked", "verified": True, "path": "script/tts-manifest.json", "segment_count": len(segments), "segmentation_plan_sha256": plan["segmentation_plan_sha256"], "created_at": now} manifest["updated_at"] = now atomic_json(manifest_path, manifest) print(json.dumps({"status": "locked", "segments": len(segments), "plan_sha256": plan["segmentation_plan_sha256"]}, ensure_ascii=False)) if __name__ == "__main__": main()