#!/usr/bin/env python3 import datetime import hashlib import json import os import subprocess import time import urllib.error import urllib.parse import urllib.request from pathlib import Path ROOT = Path('/data/video-pipeline/HaTramAudio/project/011-Hieu-Sach-Dong-Cua-Vao-Ngay-Chung-Toi-Ly-Hon') RUN_ID = 'run-20260720T040439Z-a596ad16' WORK = ROOT / 'work' / 'tts' / RUN_ID PLAN_PATH = WORK / 'chunk-plan.json' HOT_PATH = WORK / 'hot-manifest.json' ENDPOINT = 'http://192.168.40.32:7861/voice/ngoc-huyen-clone' LEASE_TOOL = Path('/home/hermes/.hermes/skills/audio-production/ha-tram-audio/scripts/gate_lease.py') def iso_now(): return datetime.datetime.now(datetime.timezone.utc).isoformat() 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): temporary = path.with_name('.' + path.name + '.tmp') with temporary.open('w', encoding='utf-8') as handle: json.dump(value, handle, ensure_ascii=False, indent=2) handle.write('\n') handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) def probe(path): output = subprocess.check_output([ 'ffprobe', '-v', 'error', '-show_entries', 'format=duration:stream=codec_name,sample_rate,channels', '-of', 'json', str(path), ]) data = json.loads(output) streams = data.get('streams', []) if len(streams) != 1 or streams[0].get('codec_name') != 'pcm_s16le': raise RuntimeError('unexpected WAV stream') if streams[0].get('sample_rate') != '48000' or streams[0].get('channels') != 1: raise RuntimeError('unexpected WAV sample format') duration = float(data['format']['duration']) if duration <= 0: raise RuntimeError('non-positive WAV duration') return data, duration def heartbeat(): subprocess.check_call([ 'python3', str(LEASE_TOOL), 'heartbeat', '--project-root', str(ROOT), '--gate', 'tts', '--run-id', RUN_ID, '--ttl-seconds', '7200', ], stdout=subprocess.DEVNULL) def request_audio(text, output_path): payload = urllib.parse.urlencode({ 'text': text, 'style': 'doc_truyen', 'speed': '1.0', 'denoise': 'true', }).encode() for attempt in range(1, 4): temporary = output_path.with_name('.' + output_path.name + f'.attempt-{attempt}.tmp') try: request = urllib.request.Request( ENDPOINT, data=payload, headers={'Content-Type': 'application/x-www-form-urlencoded'}, method='POST', ) with urllib.request.urlopen(request, timeout=600) as response: if response.status != 200: raise RuntimeError(f'HTTP {response.status}') with temporary.open('wb') as handle: while True: block = response.read(1024 * 1024) if not block: break handle.write(block) handle.flush() os.fsync(handle.fileno()) header = temporary.read_bytes()[:12] if header[:4] != b'RIFF' or header[8:12] != b'WAVE': raise RuntimeError('response is not RIFF/WAVE') probe_data, duration = probe(temporary) os.replace(temporary, output_path) return attempt, probe_data, duration except (urllib.error.URLError, TimeoutError, RuntimeError, subprocess.CalledProcessError) as exc: temporary.unlink(missing_ok=True) if attempt >= 3: raise time.sleep(2 ** attempt) def main(): lock = json.loads((ROOT / '.ownership-lock.json').read_text(encoding='utf-8')) if lock.get('owner') != 'Levy' or lock.get('run_id') != RUN_ID: raise SystemExit('ownership lock mismatch') lease = json.loads((ROOT / 'script/gate-leases/tts.json').read_text(encoding='utf-8')) if lease.get('status') != 'active' or lease.get('run_id') != RUN_ID: raise SystemExit('TTS lease mismatch') plan = json.loads(PLAN_PATH.read_text(encoding='utf-8')) narration = (ROOT / 'story/spoken-narration.txt').read_bytes() if hashlib.sha256(narration).hexdigest() != plan['spoken_narration_sha256']: raise SystemExit('narration hash mismatch') reconstructed = b''.join(Path(item['text_path']).read_bytes() for item in plan['chunks']) if reconstructed != narration: raise SystemExit('chunk reconstruction mismatch') hot = { 'schema_version': 1, 'project_id': ROOT.name, 'run_id': RUN_ID, 'status': 'running', 'plan_sha256': plan['plan_sha256'], 'spoken_narration_sha256': plan['spoken_narration_sha256'], 'started_at': iso_now(), 'chunks': [], } if HOT_PATH.exists(): previous = json.loads(HOT_PATH.read_text(encoding='utf-8')) if previous.get('plan_sha256') == plan['plan_sha256']: hot = previous hot['status'] = 'running' completed = {item['index']: item for item in hot.get('chunks', []) if item.get('status') == 'completed'} for item in plan['chunks']: index = item['index'] text_path = Path(item['text_path']) audio_path = Path(item['audio_path']) text_hash = sha256(text_path) old = completed.get(index) if old and old.get('text_sha256') == text_hash and audio_path.is_file(): try: probe_data, duration = probe(audio_path) if old.get('audio_sha256') == sha256(audio_path): heartbeat() print(f'{index:04d}/{plan["chunk_count"]} resume verified', flush=True) continue except Exception: pass attempt, probe_data, duration = request_audio(text_path.read_text(encoding='utf-8'), audio_path) result = { 'index': index, 'status': 'completed', 'word_count': item['word_count'], 'text_path': str(text_path), 'text_sha256': text_hash, 'audio_path': str(audio_path), 'audio_sha256': sha256(audio_path), 'duration_seconds': duration, 'probe': probe_data, 'attempt_count': attempt, 'completed_at': iso_now(), } completed[index] = result hot['chunks'] = [completed[key] for key in sorted(completed)] hot['completed_count'] = len(completed) hot['updated_at'] = iso_now() atomic_json(HOT_PATH, hot) heartbeat() print(f'{index:04d}/{plan["chunk_count"]} completed {duration:.3f}s attempt={attempt}', flush=True) if sorted(completed) != list(range(1, plan['chunk_count'] + 1)): raise SystemExit('segment index gap') concat_path = WORK / 'segments.ffconcat' concat_lines = ['ffconcat version 1.0'] for index in sorted(completed): escaped = completed[index]['audio_path'].replace("'", "'\\''") concat_lines.append(f"file '{escaped}'") concat_path.write_text('\n'.join(concat_lines) + '\n', encoding='utf-8') candidate = WORK / 'story-full.wav' # Keep the container extension last so FFmpeg can infer the WAV muxer. temporary = WORK / '.story-full.tmp.wav' subprocess.check_call([ 'ffmpeg', '-hide_banner', '-loglevel', 'error', '-y', '-f', 'concat', '-safe', '0', '-i', str(concat_path), '-c', 'copy', str(temporary), ]) probe_data, duration = probe(temporary) segment_total = sum(completed[index]['duration_seconds'] for index in sorted(completed)) if abs(duration - segment_total) > 0.1: raise RuntimeError(f'concat duration mismatch: {duration} vs {segment_total}') os.replace(temporary, candidate) hot.update({ 'status': 'completed', 'verified': True, 'completed_count': len(completed), 'story_full_path': str(candidate), 'story_full_sha256': sha256(candidate), 'duration_seconds': duration, 'segment_duration_total': segment_total, 'probe': probe_data, 'completed_at': iso_now(), }) atomic_json(HOT_PATH, hot) heartbeat() print(json.dumps({ 'status': 'completed', 'verified': True, 'segments': len(completed), 'duration_seconds': duration, 'story_full_sha256': hot['story_full_sha256'], }, ensure_ascii=False), flush=True) if __name__ == '__main__': main()