#!/usr/bin/env python3 from __future__ import annotations import datetime as dt import hashlib import json import os from pathlib import Path import subprocess import sys ROOT = Path("/data/video-pipeline/HaTramAudio/project/016-Nguoi-Giu-Am-Thanh-Cuoi-Cung") PROJECT = "016-Nguoi-Giu-Am-Thanh-Cuoi-Cung" RUN = "run-20260720T233725Z-24504f17" OWNER = "zoro" CANON_SHA = "fae6fce617a14dc4f8b3bd52968cc688a456c45a60bc2ed7701ccd8164875e02" ENDPOINT = "http://192.168.40.32:7861/voice/ngoc-huyen-clone" NARRATION = ROOT / "story/spoken-narration.txt" AUDIO_DIR = ROOT / "audio" ATTEMPT_DIR = ROOT / "log/tts-attempts" TMP_DIR = ROOT / "work/tts/tmp" HOT = ROOT / "work/tts/hot-manifest.json" FINAL_MANIFEST = ROOT / "script/tts-manifest.json" OUTPUT = AUDIO_DIR / "story-full.wav" REQUEST_CONFIG = {"style": "doc_truyen", "speed": "1.0", "denoise": "true"} def now() -> str: return dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z") def sha256(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: while block := f.read(8 * 1024 * 1024): h.update(block) return h.hexdigest() def atomic_json(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") os.replace(tmp, path) def guard() -> None: lock = json.loads((ROOT / ".ownership-lock.json").read_text(encoding="utf-8")) expected = (PROJECT, RUN, OWNER, "main_session_pipeline") actual = (lock.get("project_id"), lock.get("run_id"), lock.get("owner"), lock.get("status")) if actual != expected: raise RuntimeError(f"ownership mismatch: {actual!r}") resolved = ROOT.resolve() for path in (NARRATION, AUDIO_DIR, ATTEMPT_DIR, TMP_DIR, HOT, FINAL_MANIFEST, OUTPUT): if resolved not in path.resolve().parents and path.resolve() != resolved: raise RuntimeError(f"path escaped project: {path}") def chunks(text: str, target: int = 2300, hard_max: int = 3300) -> list[str]: paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()] result: list[str] = [] current: list[str] = [] size = 0 for paragraph in paragraphs: addition = len(paragraph) + (2 if current else 0) if current and size + addition > target: result.append("\n\n".join(current)) current, size = [], 0 if len(paragraph) > hard_max: raise RuntimeError("paragraph exceeds TTS hard maximum") current.append(paragraph) size += addition if current: result.append("\n\n".join(current)) if any(len(item) > hard_max for item in result): raise RuntimeError("chunk exceeds TTS hard maximum") return result def probe(path: Path) -> dict: result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration:stream=codec_name,sample_rate,channels", "-of", "json", str(path)], text=True, capture_output=True, ) if result.returncode != 0: raise RuntimeError(f"ffprobe failed for {path.name}: {result.stderr[-500:]}") data = json.loads(result.stdout) streams = data.get("streams", []) if len(streams) != 1 or not streams[0].get("codec_name"): raise RuntimeError(f"invalid audio stream for {path.name}") duration = float(data.get("format", {}).get("duration", 0)) if duration <= 0: raise RuntimeError(f"invalid duration for {path.name}") return { "duration": duration, "codec_name": streams[0].get("codec_name"), "sample_rate": int(streams[0].get("sample_rate") or 0), "channels": int(streams[0].get("channels") or 0), } def valid_checkpoint(entry: dict, text_sha: str, path: Path) -> dict | None: if entry.get("status") != "completed" or entry.get("text_sha256") != text_sha or not path.exists(): return None try: p = probe(path) except Exception: return None digest = sha256(path) if entry.get("audio_sha256") and entry["audio_sha256"] != digest: return None return {**entry, **p, "bytes": path.stat().st_size, "audio_sha256": digest, "verified": True} def main() -> None: for directory in (AUDIO_DIR, ATTEMPT_DIR, TMP_DIR, HOT.parent, FINAL_MANIFEST.parent): directory.mkdir(parents=True, exist_ok=True) test = directory / ".write-test" test.write_text("ok", encoding="ascii") test.unlink() guard() narration_sha = sha256(NARRATION) if narration_sha != CANON_SHA: raise RuntimeError(f"narration authority drift: {narration_sha}") story_manifest = json.loads((ROOT / "script/story-manifest.json").read_text(encoding="utf-8")) if story_manifest.get("canon_sha256") != CANON_SHA or story_manifest.get("spoken_narration_sha256") != CANON_SHA: raise RuntimeError("story manifest authority mismatch") texts = chunks(NARRATION.read_text(encoding="utf-8")) text_hashes = [hashlib.sha256(text.encode("utf-8")).hexdigest() for text in texts] config_hash = hashlib.sha256(json.dumps(REQUEST_CONFIG, sort_keys=True).encode()).hexdigest() old = {} if HOT.exists(): old = json.loads(HOT.read_text(encoding="utf-8")) if old.get("spoken_narration_sha256") != narration_sha or old.get("chunk_text_sha256") != text_hashes: raise RuntimeError("existing TTS checkpoint does not match narration/chunking") entries_by_index = {int(item["index"]): item for item in old.get("segments", [])} manifest = { "schema_version": 1, "project_id": PROJECT, "run_id": RUN, "status": "running", "verified": False, "endpoint": ENDPOINT, "source_canon_sha256": CANON_SHA, "spoken_narration_sha256": narration_sha, "candidate_sha256": story_manifest.get("candidate_sha256"), "request_config": REQUEST_CONFIG, "request_config_sha256": config_hash, "segment_count_planned": len(texts), "chunk_text_sha256": text_hashes, "hot_manifest_path": str(HOT), "final_manifest_path": str(FINAL_MANIFEST), "output": str(OUTPUT), "segments": [], "updated_at": now(), } atomic_json(HOT, manifest) try: for index, (text, text_sha) in enumerate(zip(texts, text_hashes), start=1): guard() output = AUDIO_DIR / f"segment-{index:04d}.wav" prior = valid_checkpoint(entries_by_index.get(index, {}), text_sha, output) if prior: manifest["segments"].append(prior) manifest["updated_at"] = now() atomic_json(HOT, manifest) continue text_path = TMP_DIR / f"segment-{index:04d}.txt" temp_output = TMP_DIR / f"segment-{index:04d}.wav.part" text_path.write_text(text, encoding="utf-8") if temp_output.exists(): temp_output.unlink() attempt = { "schema_version": 1, "project_id": PROJECT, "run_id": RUN, "status": "requesting", "verified": False, "index": index, "candidate_sha256": story_manifest.get("candidate_sha256"), "source_canon_sha256": CANON_SHA, "spoken_narration_sha256": narration_sha, "text_sha256": text_sha, "request_config": REQUEST_CONFIG, "request_config_sha256": config_hash, "output": str(output), "created_at": now(), } attempt_path = ATTEMPT_DIR / f"segment-{index:04d}.json" atomic_json(attempt_path, attempt) result = subprocess.run( [ "curl", "--fail-with-body", "--silent", "--show-error", "--max-time", "900", "-X", "POST", ENDPOINT, "-H", "Content-Type: application/x-www-form-urlencoded", "--data-urlencode", f"text@{text_path}", "--data-urlencode", "style=doc_truyen", "--data-urlencode", "speed=1.0", "--data-urlencode", "denoise=true", "--output", str(temp_output), ], text=True, capture_output=True, ) if result.returncode != 0: attempt.update(status="failed", error_type="curl", error=(result.stderr or result.stdout)[-1000:], updated_at=now()) atomic_json(attempt_path, attempt) raise RuntimeError(f"TTS request failed at segment {index}: {(result.stderr or result.stdout)[-300:]}") p = probe(temp_output) os.replace(temp_output, output) entry = { "index": index, "status": "completed", "verified": True, "text_sha256": text_sha, "output": str(output), "audio_sha256": sha256(output), "hash_method": "sha256_stream_8MiB", "bytes": output.stat().st_size, **p, } attempt.update(entry) attempt["updated_at"] = now() atomic_json(attempt_path, attempt) manifest["segments"].append(entry) manifest["updated_at"] = now() atomic_json(HOT, manifest) if len(manifest["segments"]) != len(texts): raise RuntimeError("segment count mismatch before concat") for entry, expected_hash in zip(manifest["segments"], text_hashes): checked = valid_checkpoint(entry, expected_hash, Path(entry["output"])) if not checked: raise RuntimeError(f"segment reconciliation failed: {entry.get('index')}") entry.update(checked) concat_file = TMP_DIR / "concat.txt" concat_file.write_text("".join(f"file '{Path(item['output']).as_posix()}'\n" for item in manifest["segments"]), encoding="utf-8") temp_full = TMP_DIR / "story-full.wav.part" if temp_full.exists(): temp_full.unlink() result = subprocess.run( ["ffmpeg", "-v", "error", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_file), "-ar", "48000", "-ac", "1", "-c:a", "pcm_s16le", "-f", "wav", str(temp_full)], text=True, capture_output=True, ) if result.returncode != 0: raise RuntimeError(f"concat failed: {result.stderr[-1000:]}") full_probe = probe(temp_full) segment_duration = sum(float(item["duration"]) for item in manifest["segments"]) if abs(full_probe["duration"] - segment_duration) > 0.1: raise RuntimeError(f"concat duration mismatch: {full_probe['duration']} vs {segment_duration}") os.replace(temp_full, OUTPUT) final = { **manifest, "status": "completed", "verified": True, "segment_count": len(manifest["segments"]), "output_duration": full_probe["duration"], "segment_duration_sum": segment_duration, "output_sha256": sha256(OUTPUT), "output_bytes": OUTPUT.stat().st_size, "output_probe": full_probe, "hash_method": "sha256_stream_8MiB", "completed_at": now(), "updated_at": now(), } atomic_json(HOT, final) atomic_json(FINAL_MANIFEST, final) print(json.dumps({"status": "completed", "segments": len(texts), "duration": full_probe["duration"], "output_sha256": final["output_sha256"]})) except Exception as exc: manifest.update(status="failed", verified=False, error_type=type(exc).__name__, error=str(exc), failed_segment_index=len(manifest.get("segments", [])) + 1, updated_at=now()) atomic_json(HOT, manifest) raise if __name__ == "__main__": main()