#!/usr/bin/env python3 import json, mimetypes, subprocess, time, urllib.request, uuid from pathlib import Path PROJECT = Path('/data/video-pipeline/Youtube-Video-Maker/Project/05062026-003-Hai-Ba-Trung') TTS_ENDPOINT = 'http://192.168.1.201:3900/generate' RENDER_ENDPOINT = 'http://192.168.1.104:8015/render-story-video' PROBE_ENDPOINT = 'http://192.168.1.104:8002/probe' WORKER_HOST='192.168.1.104'; WORKER_PASS='11210119' 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') LOGO=Path('/data/video-pipeline/Youtube-Video-Maker/data/logo/logoframe.png') THUMB=PROJECT/'output/thumbnails/thumb_16x9_final.png' INTRO_DIR=PROJECT/'output/intro' SEG_DIR=INTRO_DIR/'segments' INTRO_TEXT=INTRO_DIR/'intro_text.txt' INTRO_AUDIO=INTRO_DIR/'intro_audio.wav' INTRO_VIDEO=INTRO_DIR/'intro.mp4' INTRO_PROBE=INTRO_DIR/'intro_probe.json' MANIFEST=PROJECT/'logs/step17_create_intro_manifest.json' SPEED='0.82' SEGMENT_GAP_SEC=0.65 FINAL_SILENCE_SEC=3.0 def intro_segments(): scenes_doc = json.loads((PROJECT/'script/scenes.json').read_text(encoding='utf-8')) metadata = scenes_doc.get('metadata') or {} series = (metadata.get('series') or scenes_doc.get('series') or 'Việt Sử Kể Lại').lower() title = (metadata.get('title') or scenes_doc.get('title') or PROJECT.name).lower() return [ 'chào mừng quý vị trở lại với vết mực thời gian.', 'nơi quá khứ được kể lại bằng màu mực và ký ức.', f'hôm nay, trong series {series},', f'chúng ta cùng đến với {title}.', ] def log(msg): print(time.strftime('%Y-%m-%d %H:%M:%S ')+msg, flush=True) 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 tts_payload(text): return text.replace(':', ',').replace(';', ',').replace(' - ', ', ').replace('-', ',').replace('–', ',').replace('—', ',') def generate(text, out): if out.exists() and out.stat().st_size > 1000: return fields={'text':tts_payload(text),'language':'vi','ref_text':REF_TEXT.read_text(encoding='utf-8') if REF_TEXT.exists() else '', 'speed':SPEED,'num_step':'16','guidance_scale':'2.0','effect_preset':'broadcast'} last = None for attempt in range(1, 4): try: b,body=multipart(fields, {'ref_audio':REF_AUDIO}) req=urllib.request.Request(TTS_ENDPOINT,data=body,headers={'Content-Type':'multipart/form-data; boundary='+b},method='POST') with urllib.request.urlopen(req,timeout=900) as r: out.write_bytes(r.read()) return except Exception as e: last = e log(f'ERROR intro tts attempt={attempt} {e}') time.sleep(10 * attempt) raise last def post_json(url,payload,timeout=3600): req=urllib.request.Request(url,data=json.dumps(payload,ensure_ascii=False).encode(),headers={'Content-Type':'application/json'},method='POST') with urllib.request.urlopen(req,timeout=timeout) as r: return json.loads(r.read().decode()) def probe(path): d=post_json(PROBE_ENDPOINT, {'input_path':str(path)}, timeout=120); fmt=d.get('format',{}); streams=d.get('streams',[]) return {'duration_sec':float(fmt.get('duration') or 0),'format_name':fmt.get('format_name'),'size':int(fmt.get('size') or 0),'has_audio':any(s.get('codec_type')=='audio' for s in streams),'has_video':any(s.get('codec_type')=='video' for s in streams),'streams':streams} def worker(cmd,timeout=600): full=['sshpass','-p',WORKER_PASS,'ssh','-o','StrictHostKeyChecking=no','-o','PreferredAuthentications=password','-o','PubkeyAuthentication=no',f'root@{WORKER_HOST}',cmd] p=subprocess.run(full,capture_output=True,text=True,timeout=timeout) if p.returncode: raise RuntimeError(p.stderr[-4000:] or p.stdout[-4000:]) def compose_audio(paths): inputs=[]; labels=[] for i,p in enumerate(paths): inputs += ['-i', str(p)] labels.append(f'[{i}:a]aformat=sample_rates=22050:channel_layouts=mono[a{i}]') cmd = ['ffmpeg','-y','-hide_banner','-loglevel','error'] + inputs # Build silence sources directly in the filter so pauses are exact and not left to punctuation handling. filter_parts = labels[:] concat_labels=[] for i in range(len(paths)): concat_labels.append(f'[a{i}]') if i < len(paths)-1: filter_parts.append(f'anullsrc=r=22050:cl=mono:d={SEGMENT_GAP_SEC}[s{i}]') concat_labels.append(f'[s{i}]') filter_parts.append(f'anullsrc=r=22050:cl=mono:d={FINAL_SILENCE_SEC}[sf]') concat_labels.append('[sf]') filter_parts.append(''.join(concat_labels)+f'concat=n={len(concat_labels)}:v=0:a=1[outa]') cmd += ['-filter_complex',';'.join(filter_parts),'-map','[outa]','-acodec','pcm_s16le',str(INTRO_AUDIO)] # run on worker to avoid local audio processing # paths are shared, so execute the full command on the worker. quoted=' '.join("'"+x.replace("'","'\\''")+"'" for x in cmd) worker(quoted, timeout=300) def render(audio_dur): scene1=6.0; scene2=max(1.0, audio_dur-scene1) payload={'job_id':f'{PROJECT.name}_step17_intro_segmented','scenes':[{'image_path':str(LOGO),'duration':scene1},{'image_path':str(THUMB),'duration':scene2}], 'audio_path':str(INTRO_AUDIO),'subtitle_path':None,'output_path':str(INTRO_VIDEO),'width':1920,'height':1080,'fps':30,'crf':20,'preset':'veryfast','burn_subtitles':False,'ken_burns':True,'motion_preset':'none','transition':'cut','transition_duration':0.0,'fade_in':0.0,'fade_out':0.0,'slice_audio':False,'slice_subtitles':False} return post_json(RENDER_ENDPOINT,payload,timeout=3600), {'scene1_duration':scene1,'scene2_duration':scene2} def main(): INTRO_DIR.mkdir(parents=True,exist_ok=True); SEG_DIR.mkdir(parents=True,exist_ok=True); MANIFEST.parent.mkdir(parents=True,exist_ok=True) segments = intro_segments() INTRO_TEXT.write_text('\n\n'.join(segments)+'\n',encoding='utf-8') paths=[]; seg_probes=[] for i,text in enumerate(segments,1): out=SEG_DIR/f'intro_segment_{i:02d}.wav'; log(f'TTS segment {i}/{len(segments)}') generate(text,out); paths.append(out); seg_probes.append({'index':i,'text':text,'path':str(out),'probe':probe(out)}) log('compose segmented intro audio') compose_audio(paths); audio_probe=probe(INTRO_AUDIO) log(f'render segmented intro duration={audio_probe["duration_sec"]}') resp,scene_plan=render(audio_probe['duration_sec']); video_probe=probe(INTRO_VIDEO) valid=INTRO_AUDIO.exists() and INTRO_VIDEO.exists() and video_probe['has_audio'] and video_probe['has_video'] and abs(video_probe['duration_sec']-audio_probe['duration_sec'])<=0.5 doc={'project_id':PROJECT.name,'step':17,'status':'completed' if valid else 'failed_validation','mode':'segmented_intro_tts','segments':seg_probes,'segment_gap_sec':SEGMENT_GAP_SEC,'final_silence_sec':FINAL_SILENCE_SEC,'intro_text':'\n\n'.join(segments),'intro_text_path':str(INTRO_TEXT),'intro_audio_path':str(INTRO_AUDIO),'intro_video_path':str(INTRO_VIDEO),'intro_audio_probe':audio_probe,'intro_video_probe':video_probe,'render_response':resp,'scene_plan':scene_plan,'validation':{'valid':valid,'segmented_voice':True,'has_audio':video_probe['has_audio'],'has_video':video_probe['has_video']}} INTRO_PROBE.write_text(json.dumps(doc,ensure_ascii=False,indent=2),encoding='utf-8') MANIFEST.write_text(json.dumps({'project_id':PROJECT.name,'step':17,'status':doc['status'],'mode':'segmented_intro_tts','intro_audio_path':str(INTRO_AUDIO),'intro_video_path':str(INTRO_VIDEO),'intro_probe_path':str(INTRO_PROBE),'duration_sec':video_probe['duration_sec']},ensure_ascii=False,indent=2),encoding='utf-8') if not valid: raise SystemExit('failed_validation') log('DONE '+json.dumps({'status':doc['status'],'duration':video_probe['duration_sec'],'segments':len(segments)},ensure_ascii=False)) if __name__=='__main__': main()