#!/usr/bin/env python3 import json, mimetypes, re, shutil, subprocess, time, uuid, urllib.request, wave from pathlib import Path PROJECT = Path('/data/video-pipeline/Youtube-Video-Maker/Project/01062026-001-Pham-Ngu-Lao') 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') AUDIO_DIR = PROJECT / 'audio/voice_segments' TMP_DIR = PROJECT / 'audio/_targeted_tts_fix_chunks' LOG = PROJECT / 'logs/targeted_fix_3_cropped_tts_segments.log' SAMPLE_RATE = 24000 CHANNELS = 1 SAMPLE_WIDTH = 2 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} TARGETS = ['scene_020_seg_001', 'scene_027_seg_001', 'scene_081_seg_001'] 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 duration(path): 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()) 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 normalize(text): def repl(m): val = ROMAN.get(m.group(2).upper()) return f'{m.group(1)} {val}' if val else m.group(0) text = re.sub(r'(?i)\b(thế\s*k[ỷỉ])\s+([IVXLCDM]{1,6})\b', repl, text) text = re.sub(r'\s*[:;\-–—]\s*', ', ', text) text = re.sub(r',\s+', ', ', text) return text.lower().strip() def multipart(fields, files): boundary = '----OmniVoiceBoundary' + 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(): path = Path(path) mime = mimetypes.guess_type(str(path))[0] or 'application/octet-stream' chunks.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"; filename="{path.name}"\r\nContent-Type: {mime}\r\n\r\n'.encode() + path.read_bytes() + b'\r\n') chunks.append(f'--{boundary}--\r\n'.encode()) return boundary, b''.join(chunks) def postprocess(path): tmp = path.with_suffix('.clean.tmp.wav') af = 'silenceremove=start_periods=1:start_duration=0.03:start_threshold=-38dB:stop_periods=1:stop_duration=0.08:stop_threshold=-38dB,afade=t=in:st=0:d=0.015,areverse,afade=t=in:st=0:d=0.015,areverse,apad=pad_dur=0.05,adelay=250:all=1' subprocess.check_call(['ffmpeg','-y','-v','error','-i',str(path),'-af',af,str(tmp)], timeout=60) tmp.replace(path) def word_count(text): return len(re.findall(r'\w+', text, flags=re.U)) def split_chunks(text): # Make shorter requests to avoid OmniVoice returning a cropped clip for long comma-heavy sentences. if 'Quân Trần từ nhiều phía đánh ra.' in text: return ['Quân Trần từ nhiều phía đánh ra.', 'Lửa cháy, tiếng quân reo, thuyền giặc vỡ nát trên dòng nước.'] phrases = [p.strip() for p in re.split(r'(,)', text) if p.strip()] units = [] buf = '' for p in phrases: if p == ',': buf += ',' continue unit = (buf + ' ' + p).strip() if buf else p units.append(unit) buf = '' chunks, cur = [], '' for unit in units: candidate = (cur + ' ' + unit).strip() if cur else unit if cur and word_count(candidate) > 18: chunks.append(cur.rstrip(',') + '.') cur = unit else: cur = candidate if cur: chunks.append(cur if cur.endswith(('.', '!', '?', '…')) else cur.rstrip(',') + '.') return chunks def generate_chunk(text, out): ref_text = REF_TEXT.read_text(encoding='utf-8') if REF_TEXT.exists() else '' fields = {'text': normalize(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, 5): try: boundary, body = multipart(fields, files) req = urllib.request.Request(ENDPOINT, data=body, headers={'Content-Type':'multipart/form-data; boundary=' + boundary}, method='POST') with urllib.request.urlopen(req, timeout=180) as response: blob = response.read() tmp = out.with_suffix('.tmp') tmp.write_bytes(blob) if probe_fmt(tmp) != 'wav': raise RuntimeError(f'unexpected audio format first={blob[:40]!r}') tmp.replace(out) postprocess(out) dur = duration(out) if dur < 0.45: out.unlink(missing_ok=True) raise RuntimeError(f'chunk_audio_too_short duration={dur:.3f}') log(f'OK chunk {out.name} dur={dur:.3f} text={text!r}') return dur except Exception as exc: last = str(exc) log(f'ERROR chunk {out.name} attempt={attempt} {last}') time.sleep(4 * attempt) raise RuntimeError(last) def silence_frames(sec): frames = int(round(sec * SAMPLE_RATE)) return b'\x00' * frames * CHANNELS * SAMPLE_WIDTH def convert(src, dst): subprocess.check_call(['ffmpeg','-y','-v','error','-i',str(src),'-ac',str(CHANNELS),'-ar',str(SAMPLE_RATE),'-sample_fmt','s16',str(dst)], timeout=60) def concat_chunks(chunk_paths, output): norm_paths = [] for idx, chunk in enumerate(chunk_paths, 1): norm = chunk.with_name(chunk.stem + '_norm.wav') convert(chunk, norm) norm_paths.append(norm) output.parent.mkdir(parents=True, exist_ok=True) with wave.open(str(output), 'wb') as out: out.setnchannels(CHANNELS) out.setsampwidth(SAMPLE_WIDTH) out.setframerate(SAMPLE_RATE) for idx, path in enumerate(norm_paths): with wave.open(str(path), 'rb') as inp: out.writeframes(inp.readframes(inp.getnframes())) if idx != len(norm_paths) - 1: out.writeframes(silence_frames(0.12)) def main(): LOG.write_text('', encoding='utf-8') TMP_DIR.mkdir(parents=True, exist_ok=True) doc = json.loads((AUDIO_DIR / 'voice_segments_manifest.json').read_text(encoding='utf-8')) lookup = {item['voice_segment_id']: item for item in doc['items']} stamp = time.strftime('%Y%m%d_%H%M%S') backup = PROJECT / 'audio/voice_segments_backups' / f'targeted_chunk_fix_{stamp}' backup.mkdir(parents=True, exist_ok=True) results = [] for voice_id in TARGETS: item = lookup[voice_id] out = Path(item['audio_path']) if out.exists(): dst = backup / item['scene_id'] / out.name dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(out, dst) out.unlink() chunks = split_chunks(item['text']) log(f'START {voice_id} chunks={len(chunks)}') chunk_paths = [] for idx, chunk_text in enumerate(chunks, 1): chunk_out = TMP_DIR / voice_id / f'chunk_{idx:02d}.wav' chunk_out.parent.mkdir(parents=True, exist_ok=True) if chunk_out.exists(): chunk_out.unlink() generate_chunk(chunk_text, chunk_out) chunk_paths.append(chunk_out) concat_chunks(chunk_paths, out) final_duration = duration(out) results.append({'voice_segment_id': voice_id, 'output': str(out), 'chunks': chunks, 'duration': final_duration, 'bytes': out.stat().st_size}) log(f'DONE_SEG {voice_id} dur={final_duration:.3f} bytes={out.stat().st_size}') manifest = {'status':'completed', 'targets': results, 'backup_dir': str(backup)} (PROJECT / 'logs/targeted_fix_3_cropped_tts_segments_manifest.json').write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding='utf-8') print(json.dumps(manifest, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()