#!/usr/bin/env python3 from pathlib import Path import datetime as dt, hashlib, json, re, subprocess, urllib.parse, urllib.request, wave ROOT = Path('/data/video-pipeline/HaTramAudio/project/021-Ngay-Toi-Bi-Gach-Ten-Khoi-Gia-Pha') PROJECT = ROOT.name RUN = 'run-20260721T131911Z-ac5d64d1' CANON = 'bb80e97d9adcb0427ea3828449858445323db228cca38f37cdf0446fffefdd6f' ENDPOINT = 'http://192.168.40.32:7861/voice/ngoc-huyen-clone' NARRATION = ROOT / 'script/spoken-narration.txt' OUTDIR = ROOT / 'audio/chunks' HOT = ROOT / 'work/tts/hot-manifest.json' FINAL = ROOT / 'script/tts-manifest.json' FULL = ROOT / 'audio/story-full.wav' CONFIG = {'style': 'doc_truyen', 'speed': '1.0', 'denoise': 'true'} def now(): return dt.datetime.now(dt.timezone.utc).isoformat().replace('+00:00', 'Z') def sha(path): h = hashlib.sha256() with open(path, 'rb') as f: while block := f.read(8 * 1024 * 1024): h.update(block) return h.hexdigest() def atomic_json(path, value): 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') tmp.replace(path) def guard(): lock = json.loads((ROOT / '.ownership-lock.json').read_text(encoding='utf-8')) assert lock['project_id'] == PROJECT and lock['run_id'] == RUN assert lock['owner'] == 'zoro' and lock['status'] == 'main_session_pipeline' assert sha(NARRATION) == CANON story = json.loads((ROOT / 'script/story-manifest.json').read_text(encoding='utf-8')) assert story['verified'] and story['spoken_narration_sha256'] == CANON def split_chunks(text, limit=2700): paragraphs = [x.strip() for x in text.split('\n\n') if x.strip()] units = [] for paragraph in paragraphs: if len(paragraph) <= limit: units.append(paragraph) continue sentences = re.split(r'(?<=[.!?…”])\s+', paragraph) current = '' for sentence in sentences: if len(sentence) > limit: raise RuntimeError('sentence exceeds chunk limit') trial = sentence if not current else current + ' ' + sentence if len(trial) > limit: units.append(current) current = sentence else: current = trial if current: units.append(current) chunks, current = [], '' for unit in units: trial = unit if not current else current + '\n\n' + unit if len(trial) > limit: chunks.append(current) current = unit else: current = trial if current: chunks.append(current) assert ''.join(re.sub(r'\s+', '', x) for x in chunks) == re.sub(r'\s+', '', text) return chunks def probe_wav(path): with wave.open(str(path), 'rb') as w: rate, channels, width, frames = w.getframerate(), w.getnchannels(), w.getsampwidth(), w.getnframes() duration = frames / rate assert rate > 0 and channels > 0 and width > 0 and frames > 0 and duration > 0 return {'sample_rate': rate, 'channels': channels, 'sample_width': width, 'frames': frames, 'duration_seconds': duration} def main(): guard() OUTDIR.mkdir(parents=True, exist_ok=True) text = NARRATION.read_text(encoding='utf-8') chunks = split_chunks(text) manifest = {'schema_version': 1, 'project_id': PROJECT, 'run_id': RUN, 'status': 'running', 'verified': False, 'source_canon_sha256': CANON, 'spoken_narration_sha256': CANON, 'endpoint': ENDPOINT, 'request_config': CONFIG, 'chunk_count': len(chunks), 'segments': [], 'updated_at': now()} if HOT.exists(): old = json.loads(HOT.read_text(encoding='utf-8')) if old.get('source_canon_sha256') == CANON and old.get('chunk_count') == len(chunks): manifest['segments'] = old.get('segments', []) completed = {x['index']: x for x in manifest['segments'] if x.get('status') == 'completed'} for index, chunk in enumerate(chunks, 1): guard() out = OUTDIR / f'segment-{index:04d}.wav' text_hash = hashlib.sha256(chunk.encode('utf-8')).hexdigest() prior = completed.get(index) if prior and prior.get('text_sha256') == text_hash and out.exists(): probe = probe_wav(out) if prior.get('audio_sha256') == sha(out) and abs(prior.get('duration_seconds', 0) - probe['duration_seconds']) < 0.001: continue attempt = {'schema_version': 1, 'project_id': PROJECT, 'run_id': RUN, 'status': 'creating', 'candidate_sha256': CANON, 'source_canon_sha256': CANON, 'spoken_narration_sha256': CANON, 'index': index, 'text_sha256': text_hash, 'request_config': CONFIG, 'output_path': str(out), 'created_at': now()} atomic_json(ROOT / f'log/tts-attempt-{index:04d}.json', attempt) data = urllib.parse.urlencode({'text': chunk, **CONFIG}).encode('utf-8') request = urllib.request.Request(ENDPOINT, data=data, headers={'Content-Type': 'application/x-www-form-urlencoded'}, method='POST') temp = out.with_suffix('.wav.part') try: with urllib.request.urlopen(request, timeout=900) as response, open(temp, 'wb') as f: if response.status != 200: raise RuntimeError(f'HTTP {response.status}') while block := response.read(1024 * 1024): f.write(block) if temp.stat().st_size == 0: raise RuntimeError('empty WAV response') temp.replace(out) probe = probe_wav(out) segment = {'index': index, 'status': 'completed', 'text_sha256': text_hash, 'output_path': str(out), 'size_bytes': out.stat().st_size, 'audio_sha256': sha(out), 'hash_method': 'sha256_stream_8MiB', 'duration_seconds': probe['duration_seconds'], 'probe': probe, 'completed_at': now()} manifest['segments'] = [x for x in manifest['segments'] if x.get('index') != index] + [segment] manifest['segments'].sort(key=lambda x: x['index']) manifest['updated_at'] = now() atomic_json(HOT, manifest) attempt.update({'status': 'completed', 'audio_sha256': segment['audio_sha256'], 'duration_seconds': segment['duration_seconds'], 'completed_at': now()}) atomic_json(ROOT / f'log/tts-attempt-{index:04d}.json', attempt) except Exception as exc: if temp.exists(): temp.unlink() manifest.update({'status': 'failed', 'verified': False, 'failed_segment_index': index, 'failure_type': type(exc).__name__, 'failure': str(exc), 'updated_at': now()}) atomic_json(HOT, manifest) attempt.update({'status': 'failed', 'failure_type': type(exc).__name__, 'failure': str(exc), 'failed_at': now()}) atomic_json(ROOT / f'log/tts-attempt-{index:04d}.json', attempt) raise segments = sorted(manifest['segments'], key=lambda x: x['index']) assert [x['index'] for x in segments] == list(range(1, len(chunks) + 1)) for item, chunk in zip(segments, chunks): out = Path(item['output_path']) assert item['text_sha256'] == hashlib.sha256(chunk.encode('utf-8')).hexdigest() assert item['audio_sha256'] == sha(out) probe_wav(out) concat = ROOT / 'work/tts/concat.txt' concat.write_text(''.join("file '" + str(Path(x['output_path'])).replace("'", "'\\''") + "'\n" for x in segments), encoding='utf-8') subprocess.run(['ffmpeg', '-hide_banner', '-loglevel', 'error', '-y', '-f', 'concat', '-safe', '0', '-i', str(concat), '-ac', '1', '-ar', '48000', '-c:a', 'pcm_s16le', str(FULL)], check=True) full_probe = probe_wav(FULL) manifest.update({'status': 'completed', 'verified': True, 'segment_count': len(segments), 'output': str(FULL), 'output_size_bytes': FULL.stat().st_size, 'output_duration': full_probe['duration_seconds'], 'output_sha256': sha(FULL), 'output_probe': full_probe, 'hash_method': 'sha256_stream_8MiB', 'completed_at': now()}) atomic_json(HOT, manifest) atomic_json(FINAL, manifest) print(json.dumps({'status': 'completed', 'verified': True, 'segments': len(segments), 'duration_seconds': full_probe['duration_seconds'], 'bytes': FULL.stat().st_size, 'sha256': manifest['output_sha256']}, ensure_ascii=False)) if __name__ == '__main__': main()