#!/usr/bin/env python3 import concurrent.futures, json, mimetypes, re, time, uuid, urllib.request, subprocess from pathlib import Path PROJECT=Path('/data/video-pipeline/Youtube-Video-Maker/Project/11062026-002-Nha-Dinh') ENDPOINT='http://192.168.1.201:3900/generate' REF_AUDIO=Path('/data/video-pipeline/Youtube-Video-Maker/data/voices/template1/refaudio.mp3') REF_TEXT=Path('/data/video-pipeline/Youtube-Video-Maker/data/voices/template1/reftext.txt') WORKERS=1 LEADING_SILENCE_SEC=0.25 TRAILING_SILENCE_SEC=0.35 MIN_TTS_WORDS=8 MIN_TTS_CHARS=32 EDGE_TRIM_DB=-38 FADE_SEC=0.015 AUDIO_DIR=PROJECT/'audio/voice_segments' LEGACY_AUDIO_DIR=PROJECT/'audio/scenes' LOG=PROJECT/'logs/step_07_tts_v2_run.log' MANIFEST=PROJECT/'logs/step_07_tts_v2_manifest.json' TEXT_MANIFEST=PROJECT/'audio/voice_segments/voice_segments_manifest.json' INSTRUCT=('Đọc giọng kể chuyện lịch sử Việt Nam, trầm, chậm rãi, rõ chữ. ' 'Mỗi đoạn là một câu hoặc cụm ngắt mạnh, không đọc vội, giữ nhịp kể chuyện.') 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 split_voice_segments(text): parts=re.findall(r'[^.!?…:]+[.!?…:]+(?:[”"])?|[^.!?…:]+$', text.strip(), flags=re.U) parts=[re.sub(r'\s+', ' ', p).strip() for p in parts if p.strip()] return merge_short_segments(parts) def word_count(text): return len(re.findall(r'\w+', text, flags=re.U)) def is_short_segment(text): return word_count(text) < MIN_TTS_WORDS or len(text.strip()) < MIN_TTS_CHARS def merge_short_segments(parts): if len(parts) <= 1: return parts merged=[] for part in parts: if not merged: merged.append(part) continue if is_short_segment(part): merged[-1]=(merged[-1].rstrip()+' '+part.lstrip()).strip() elif is_short_segment(merged[-1]): merged[-1]=(merged[-1].rstrip()+' '+part.lstrip()).strip() else: merged.append(part) return merged def normalize(text): changes=[] def repl(m): numeral=m.group(2).upper(); val=ROMAN.get(numeral) 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) before=text2 text2=re.sub(r'\s*[:;\-–—]\s*', ', ', text2) text2=re.sub(r'([.!?…])\s+', r'\1\n', text2) text2=re.sub(r',\s+', ', ', text2) text2=re.sub(r'\n{2,}', '\n', text2).strip() if text2!=before: changes.append('punctuation_pause_normalized_for_tts') lower=text2.lower() if lower!=text2: changes.append('lowercase_for_omnivoice') return lower, changes def multipart(fields, files): b='----OmniVoiceBoundary'+uuid.uuid4().hex chunks=[] for k,v in fields.items(): chunks.append(f'--{b}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode()) for k,path in files.items(): p=Path(path); mime=mimetypes.guess_type(str(p))[0] or 'application/octet-stream' chunks.append(f'--{b}\r\nContent-Disposition: form-data; name="{k}"; filename="{p.name}"\r\nContent-Type: {mime}\r\n\r\n'.encode()+p.read_bytes()+b'\r\n') chunks.append(f'--{b}--\r\n'.encode()) return b,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') 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},areverse,' f'afade=t=in:st=0:d={FADE_SEC},areverse,' 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 generate(item): out=Path(item['audio_path']) if out.exists() and out.stat().st_size > 1000: 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':duration(out)} out.parent.mkdir(parents=True, exist_ok=True) ref_text=REF_TEXT.read_text(encoding='utf-8') if REF_TEXT.exists() else '' fields={'text':item['tts_text'],'language':'vi','ref_text':ref_text,'speed':'0.88','num_step':'16','guidance_scale':'2.0','effect_preset':'broadcast'} files={'ref_audio':REF_AUDIO} last=None for attempt in range(1,4): try: b,body=multipart(fields, files) req=urllib.request.Request(ENDPOINT,data=body,headers={'Content-Type':'multipart/form-data; boundary='+b},method='POST') with urllib.request.urlopen(req, timeout=900) as r: blob=r.read(); ct=r.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 {'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':duration(out),'normalization':item['normalization']} except Exception as e: last=str(e); log(f"ERROR {item['voice_segment_id']} attempt={attempt} {last}"); time.sleep(5*attempt) return {'scene_id':item['scene_id'],'voice_segment_id':item['voice_segment_id'],'status':'failed','error':last,'output':str(out)} def build_items(): scenes=json.loads((PROJECT/'script/scenes.json').read_text(encoding='utf-8'))['scenes'] items=[] for scene in scenes: raw_segments=[scene['narration']] if scene.get('type')=='heading' else split_voice_segments(scene['narration']) for idx,txt in enumerate(raw_segments,1): tts,changes=normalize(txt) seg_id=f"{scene['scene_id']}_seg_{idx:03d}" audio_path=AUDIO_DIR/scene['scene_id']/f'seg_{idx:03d}.wav' items.append({'scene_id':scene['scene_id'],'voice_segment_id':seg_id,'segment_index_in_scene':idx,'segments_in_scene':len(raw_segments),'scene_type':scene.get('type'),'text':txt,'source_narration':scene['narration'],'tts_text':tts,'normalization':changes,'image_path':str(PROJECT/'images/scenes'/f"{scene['scene_id']}.png"),'audio_path':str(audio_path)}) return items def main(): AUDIO_DIR.mkdir(parents=True, exist_ok=True); LEGACY_AUDIO_DIR.mkdir(parents=True, exist_ok=True) LOG.write_text('', encoding='utf-8') items=build_items() TEXT_MANIFEST.parent.mkdir(parents=True, exist_ok=True) TEXT_MANIFEST.write_text(json.dumps({'project_id':PROJECT.name,'split_rule':f'Split voice at . ! ? … and :; auto-merge very short segments below {MIN_TTS_WORDS} words or {MIN_TTS_CHARS} chars to reduce TTS edge artifacts.','instruct':INSTRUCT,'speed':0.88,'leading_silence_sec':LEADING_SILENCE_SEC,'trailing_silence_sec':TRAILING_SILENCE_SEC,'items':items},ensure_ascii=False,indent=2),encoding='utf-8') log(f'START v2 voice_segments total={len(items)} workers={WORKERS}') results=[] with concurrent.futures.ThreadPoolExecutor(max_workers=WORKERS) as ex: futs=[ex.submit(generate,it) for it in items] for fut in concurrent.futures.as_completed(futs): r=fut.result(); results.append(r); log(f"{r['status']} {r['voice_segment_id']} {r.get('bytes',0)} dur={r.get('duration')}") failed=[r for r in results if r['status']=='failed'] manifest={'project_id':PROJECT.name,'step':7,'version':'v2_voice_segments','status':'completed' if not failed else 'partial_failed','total_voice_segments':len(items),'done':len(items)-len(failed),'failed':failed,'total_duration_seconds':sum((r.get('duration') or 0) for r in results),'results':sorted(results,key=lambda x:x['voice_segment_id'])} MANIFEST.write_text(json.dumps(manifest,ensure_ascii=False,indent=2),encoding='utf-8') log('DONE '+json.dumps({'status':manifest['status'],'done':manifest['done'],'duration_min':round(manifest['total_duration_seconds']/60,2)},ensure_ascii=False)) if failed: raise SystemExit('partial_failed') if __name__=='__main__': main()