#!/usr/bin/env python3 import concurrent.futures import json import mimetypes import re import subprocess import time import urllib.request import uuid from pathlib import Path PROJECT = Path('/data/video-pipeline/Youtube-Video-Maker/Project/04072026-001-To-Vinh-Dien') ENDPOINT = 'http://192.168.40.32:7861/voice/truong-giang' TTS_STYLE = 'doc_truyen' AUDIO_DIR = PROJECT / 'audio/voice_segments' TEXT_MANIFEST = AUDIO_DIR / 'voice_segments_manifest.json' LOG = PROJECT / 'logs/step_07_tts_v2_run.log' MANIFEST = PROJECT / 'logs/step_07_tts_v2_manifest.json' SCENES_JSON = PROJECT / 'script/scenes.json' WORKERS = 1 MAX_ATTEMPTS = 3 TTS_TIMEOUT = 120 LEADING_SILENCE_SEC = 0.22 TRAILING_SILENCE_SEC = 0.08 EDGE_TRIM_DB = -38 FADE_SEC = 0.015 SPEED = '0.88' ROMAN = {'I':1,'II':2,'III':3,'IV':4,'V':5,'VI':6,'VII':7,'VIII':8,'IX':9,'X':10,'XI':11,'XII':12,'XIII':13,'XIV':14,'XV':15,'XVI':16,'XVII':17,'XVIII':18,'XIX':19,'XX':20} def log(msg): line = time.strftime('%Y-%m-%d %H:%M:%S ') + msg LOG.parent.mkdir(parents=True, exist_ok=True) with LOG.open('a', encoding='utf-8') as f: f.write(line + '\n') print(line, flush=True) def clean_spaces(text): return re.sub(r'\s+', ' ', text.strip()) def word_count(text): return len(re.findall(r'\w+', text, flags=re.U)) def split_voice_segments(text, scene_type): text = clean_spaces(text) if not text: return [] if scene_type == 'heading': return [text] parts = re.findall(r'[^.!?…]+[.!?…]+(?:[”"])?|[^.!?…]+$', text, flags=re.U) parts = [clean_spaces(p) for p in parts if clean_spaces(p)] out = [] for i, part in enumerate(parts): if word_count(part) <= 3 and len(parts) > 1: if out: out[-1] = clean_spaces(out[-1] + ' ' + part) elif i + 1 < len(parts): parts[i + 1] = clean_spaces(part + ' ' + parts[i + 1]) else: out.append(part) else: out.append(part) return out def normalize_for_tts(text): changes = [] def repl(m): val = ROMAN.get(m.group(2).upper()) if val: changes.append(f'{m.group(0)}->{m.group(1)} {val}') return f'{m.group(1)} {val}' return m.group(0) text2 = re.sub(r'(?i)\b(thế\s*k[ỷỉ])\s+([IVXLCDM]{1,6})\b', repl, text) text2 = re.sub(r'\s*[:;\-–—]\s*', ', ', text2) text2 = re.sub(r',\s+', ', ', text2) text2 = clean_spaces(text2) if text2 != text: changes.append('tts_pause_normalized') return text2, changes def multipart(fields, files): boundary = '----TruongGiangBoundary' + uuid.uuid4().hex chunks = [] for key, value in fields.items(): chunks.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n{value}\r\n'.encode()) for key, path in files.items(): p = Path(path) mime = mimetypes.guess_type(str(p))[0] or 'application/octet-stream' chunks.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"; filename="{p.name}"\r\nContent-Type: {mime}\r\n\r\n'.encode() + p.read_bytes() + b'\r\n') chunks.append(f'--{boundary}--\r\n'.encode()) return boundary, b''.join(chunks) def probe_fmt(path): data = Path(path).read_bytes()[:16] return 'wav' if data.startswith(b'RIFF') and b'WAVE' in data[:12] else 'unknown' def duration(path): try: out = subprocess.check_output(['ffprobe', '-v', 'error', '-show_entries', 'format=duration', '-of', 'default=nw=1:nk=1', str(path)], text=True, timeout=10) return float(out.strip()) except Exception: return None def postprocess_audio(path): tmp = path.with_suffix('.clean.tmp.wav') # Do not trim the tail: narration TTS often inserts natural pauses, and tail trimming can cut speech. # Only remove leading junk, add tiny fades, then pad edges for safer downstream cuts. af = ( f'silenceremove=start_periods=1:start_duration=0.03:start_threshold={EDGE_TRIM_DB}dB,' f'afade=t=in:st=0:d={FADE_SEC},' f'apad=pad_dur={TRAILING_SILENCE_SEC},' f'adelay={int(LEADING_SILENCE_SEC * 1000)}:all=1' ) subprocess.check_call(['ffmpeg', '-y', '-v', 'error', '-i', str(path), '-af', af, str(tmp)], timeout=60) tmp.replace(path) def request_tts(text, out): fields = { 'text': text, 'style': TTS_STYLE, } boundary, body = multipart(fields, {}) req = urllib.request.Request(ENDPOINT, data=body, headers={'Content-Type': 'multipart/form-data; boundary=' + boundary}, method='POST') with urllib.request.urlopen(req, timeout=TTS_TIMEOUT) as resp: blob = resp.read() ct = resp.headers.get('Content-Type', '') tmp = out.with_suffix('.tmp') tmp.write_bytes(blob) if probe_fmt(tmp) != 'wav': raise RuntimeError(f'unexpected format ct={ct} first={blob[:60]!r}') tmp.replace(out) postprocess_audio(out) return duration(out) def generate(item): out = Path(item['audio_path']) if out.exists() and out.stat().st_size > 1000: dur = duration(out) return {'scene_id': item['scene_id'], 'voice_segment_id': item['voice_segment_id'], 'status': 'skipped_exists', 'output': str(out), 'bytes': out.stat().st_size, 'duration': dur} out.parent.mkdir(parents=True, exist_ok=True) last = None for attempt in range(1, MAX_ATTEMPTS + 1): try: out.unlink(missing_ok=True) dur = request_tts(item['tts_text'], out) return {'scene_id': item['scene_id'], 'voice_segment_id': item['voice_segment_id'], 'status': 'ok', 'attempt': attempt, 'output': str(out), 'bytes': out.stat().st_size, 'duration': dur, 'normalization': item['normalization']} except Exception as exc: last = str(exc) out.unlink(missing_ok=True) log(f"ERROR {item['voice_segment_id']} attempt={attempt} {last}") time.sleep(4 * attempt) return {'scene_id': item['scene_id'], 'voice_segment_id': item['voice_segment_id'], 'status': 'failed', 'error': last, 'duration': None} def build_items(): scenes = json.loads(SCENES_JSON.read_text(encoding='utf-8'))['scenes'] items = [] for scene in scenes: sid = scene['scene_id'] stype = scene.get('type', '') narration = scene.get('narration') or scene.get('text') or '' parts = split_voice_segments(narration, stype) for idx, part in enumerate(parts, 1): tts_text, changes = normalize_for_tts(part) items.append({ 'project_id': PROJECT.name, 'scene_id': sid, 'scene_type': stype, 'voice_segment_id': f'{sid}_seg_{idx:03d}', 'segment_index_in_scene': idx, 'segments_in_scene': len(parts), 'source_narration': narration, 'text': part, 'tts_text': tts_text, 'audio_path': str(AUDIO_DIR / sid / f'seg_{idx:03d}.wav'), 'image_path': str(PROJECT / 'images/scenes' / f'{sid}.png'), 'normalization': changes, }) TEXT_MANIFEST.parent.mkdir(parents=True, exist_ok=True) TEXT_MANIFEST.write_text(json.dumps({ 'project_id': PROJECT.name, 'version': 'sentence_split_truong_giang_doc_truyen_head_trim_only', 'split_rule': 'Heading scenes stay separate. Normal narration splits only on sentence-ending punctuation (. ! ? …); 2-3 word non-heading fragments merge into previous/next segment. TTS uses Truong Giang endpoint with style=doc_truyen; head-trim-only postprocess.', 'items': items, }, ensure_ascii=False, indent=2), encoding='utf-8') return items def main(): LOG.write_text('', encoding='utf-8') items = build_items() log(f'START sentence_split_user_qa voice_segments total={len(items)} workers={WORKERS}') results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=WORKERS) as executor: futures = [executor.submit(generate, item) for item in items] for future in concurrent.futures.as_completed(futures): result = future.result() results.append(result) if result['status'] in ('ok', 'skipped_exists'): log(f"{result['status']} {result['voice_segment_id']} {result.get('bytes')} dur={result.get('duration')}") else: log(f"failed {result['voice_segment_id']} {result.get('error')}") failed = [r for r in results if r['status'] == 'failed'] manifest = { 'project_id': PROJECT.name, 'step': 7, 'version': 'sentence_split_truong_giang_doc_truyen_head_trim_only', 'status': 'completed' if not failed else 'partial_failed', 'total_voice_segments': len(items), 'done': len(items) - len(failed), 'failed': failed, 'results': results, } MANIFEST.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding='utf-8') log('DONE ' + json.dumps({'status': manifest['status'], 'done': manifest['done'], 'total': len(items)}, ensure_ascii=False)) if failed: raise SystemExit('partial_failed') if __name__ == '__main__': main()