import gradio as gr import os import re import pysrt import io import wave import json import traceback import threading import concurrent.futures import librosa import numpy as np import soundfile as sf import pyrubberband as pyrb from piper.voice import PiperVoice from vinorm import TTSnorm from huggingface_hub import hf_hub_download, list_repo_files # --- 1. QUẢN LÝ MODEL VÀ CẤU HÌNH --- REPO_ID = "hoanglinhn0/Model" voice_cache = {} model_lock = threading.Lock() CONFIG_FILE = "config.json" def load_config(): if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, "r", encoding="utf-8") as f: return json.load(f) except: return {} return {} def save_config_to_file(voice, clean_opts, vol, speed, pitch, threads, ns, nw): data = { "voice": voice, "clean_opts": clean_opts, "vol": vol, "speed": speed, "pitch": pitch, "threads": threads, "ns": ns, "nw": nw } try: with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=4) except Exception as e: print(f"Lỗi lưu config: {e}") def get_voice_list(): try: files = list_repo_files(repo_id=REPO_ID) models = [f for f in files if f.endswith('.onnx')] return sorted(models) except Exception as e: print(f"Lỗi lấy danh sách model: {e}") return [] def load_piper_voice(model_name): with model_lock: if model_name not in voice_cache: try: print(f"📥 Đang tải model: {model_name}...") onnx_path = hf_hub_download(repo_id=REPO_ID, filename=model_name) config_path = hf_hub_download(repo_id=REPO_ID, filename=model_name + ".json") voice_cache[model_name] = PiperVoice.load(onnx_path, config_path=config_path) print(f"✅ Đã tải xong: {model_name}") except Exception as e: raise Exception(f"Không thể tải model {model_name}: {str(e)}") return voice_cache[model_name] # --- 2. HÀM LÀM SẠCH VĂN BẢN --- def advanced_text_cleaning(text, options=None): if options is None: options = [] if not text: return "" if "Bỏ qua các thẻ HTML" in options: text = re.sub(r'<[^>]+>', ' ', text) if "Bỏ qua đoạn văn giữa dấu ngoặc đơn ()" in options: text = re.sub(r'\([^)]*\)', '', text) if "Bỏ qua văn bản giữa dấu xoăn {}" in options: text = re.sub(r'\{[^}]*\}', '', text) if "Lờ chữ cái giữa hai dấu vuông []" in options: text = re.sub(r'\[[^\]]*\]', '', text) if "Bỏ qua những đoạn văn giữa dấu hoa thị *" in options: text = re.sub(r'\*[^*]*\*', '', text) if "Bỏ qua văn bản giữa các ghi chú ♪ ♪" in options: text = re.sub(r'♪[^♪]*♪', '', text) text = text.replace('♪', '') if "Lờ đi các ký tự: * # ~" in options: for char in ['*', '#', '~', '.', '!', '?']: text = text.replace(char, '') try: text = TTSnorm(text) except: pass text = re.sub(r'(?i)\s+chấm([.?!]|$)', '.', text) text = re.sub(r'(?i)\s+phẩy([,;]|$)', ',', text) text = re.sub(r'\s+', ' ', text).strip() text = re.sub(r'\s+([.?!,;:])', r'\1', text) return text # --- 3. CÁC HÀM XỬ LÝ DSP TỐI ƯU HÓA --- def safe_normalize(audio_array, target_peak=0.90): audio_array = np.array(audio_array, dtype=np.float32) max_amp = np.max(np.abs(audio_array)) if max_amp > 0: audio_array = (audio_array / max_amp) * target_peak return audio_array def apply_fade(audio_array, sr, fade_duration=0.015): fade_samples = int(sr * fade_duration) if len(audio_array) < fade_samples * 2: return audio_array fade_in = np.linspace(0.0, 1.0, fade_samples, dtype=np.float32) fade_out = np.linspace(1.0, 0.0, fade_samples, dtype=np.float32) audio_array[:fade_samples] *= fade_in audio_array[-fade_samples:] *= fade_out return audio_array # --- 4. GỌI PIPER ĐỂ TẠO AUDIO (Trả về Numpy Array) --- def get_piper_audio(text, voice_name, speed_m, noise, noise_w): try: voice = load_piper_voice(voice_name) audio_stream = io.BytesIO() with wave.open(audio_stream, "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(22050) if len(text) < 300: # Ép tốc độ gốc trực tiếp trên Piper voice.synthesize(text, wav_file, length_scale=1.0/speed_m, noise_scale=noise, noise_w=noise_w) else: text_formatted = re.sub(r'([.?!])\s*', r'\1\n', text) for part in text_formatted.split('\n'): part = part.strip() if part and not re.match(r'^[.?!,;:\s]+$', part): voice.synthesize(part, wav_file, length_scale=1.0/speed_m, noise_scale=noise, noise_w=noise_w) audio_stream.seek(0) audio_array, sr = librosa.load(audio_stream, sr=22050) return audio_array, sr except Exception as e: print(f"Lỗi Piper Synth: {e}") return None, 22050 # --- 5. HÀM XỬ LÝ LÕI ĐA LUỒNG TỪNG DÒNG PHỤ ĐỀ --- def process_single_line(index, text, start_sec, end_sec, voice_name, native_sr, speed_m, pitch_steps, noise, noise_w): srt_duration = end_sec - start_sec if srt_duration <= 0: srt_duration = 0.5 # 1. Gọi Piper lấy âm thanh gốc audio_chunk, _ = get_piper_audio(text, voice_name, speed_m, noise, noise_w) if audio_chunk is None or len(audio_chunk) == 0: return index, None, None, None # 2. Xén khoảng lặng và Bơm "đệm khí" (Buffer) 100ms audio_chunk_trimmed, _ = librosa.effects.trim(audio_chunk, top_db=35) if len(audio_chunk_trimmed) > native_sr * 0.2: audio_chunk = audio_chunk_trimmed silence_samples = int(0.1 * native_sr) half_silence = np.zeros(silence_samples // 2, dtype=np.float32) audio_chunk = np.concatenate((half_silence, audio_chunk, half_silence)) audio_chunk = safe_normalize(audio_chunk, target_peak=0.90) orig_dur = len(audio_chunk) / native_sr # 3. Tính toán bù trừ thời gian theo SRT # Tốc độ đã được Piper buff, ta chỉ ép thêm nếu file vẫn dài hơn SRT stretch_factor = orig_dur / srt_duration stretch_factor = max(1.0, stretch_factor) # 4. Ép thời gian/cao độ an toàn bằng pyrubberband if abs(stretch_factor - 1.0) > 0.01 or pitch_steps != 0: try: processed_chunk = pyrb.time_stretch(audio_chunk, native_sr, stretch_factor) if pitch_steps != 0: processed_chunk = pyrb.pitch_shift(processed_chunk, native_sr, pitch_steps) except Exception as e: processed_chunk = audio_chunk print(f"Bỏ qua ép tốc độ dòng {index}: {e}") else: processed_chunk = audio_chunk processed_chunk = safe_normalize(processed_chunk, target_peak=0.90) processed_chunk = apply_fade(processed_chunk, native_sr, fade_duration=0.015) start_sample = int(start_sec * native_sr) return index, processed_chunk, start_sample, stretch_factor # --- 6. HÀM GIAO TIẾP VỚI GRADIO (PIPELINE CHÍNH) --- def process_pipeline(voice_name, srt_file, manual_text, base_speed, pitch_steps, threads_count, noise, noise_w, vol_pct, clean_options, progress=gr.Progress(track_tqdm=True)): try: log_messages = [] def log(msg): log_messages.append(msg) return "\n".join(log_messages) if not voice_name: return None, log("⚠️ Chưa chọn model!") subs = [] if srt_file: try: subs = pysrt.open(srt_file.name, encoding='utf-8') except: return None, log("⚠️ Lỗi đọc file SRT!") elif manual_text.strip(): try: temp_subs = pysrt.from_string(manual_text) if len(temp_subs) > 0 and (temp_subs[0].end.ordinal > 0 or temp_subs[-1].end.ordinal > 0): subs = temp_subs else: raise Exception() except: return None, log("⚠️ Dữ liệu không đúng chuẩn SRT. Vui lòng tải file .srt lên.") else: return None, log("⚠️ Không có dữ liệu phụ đề.") yield None, log("⏳ Đang khởi tạo model và phân tích Timeline...") # Load trước model để vào đa luồng không bị nghẽn load_piper_voice(voice_name) native_sample_rate = 22050 max_end_sec = max([sub.end.ordinal / 1000.0 for sub in subs]) total_samples = int((max_end_sec + 2.0) * native_sample_rate) # Dư ra 2 giây an toàn final_audio = np.zeros(total_samples, dtype=np.float32) yield None, log(f"🚀 Bắt đầu tổng hợp đa luồng ({threads_count} luồng) cho {len(subs)} câu...") tasks = [] for i, item in enumerate(subs): clean_text = advanced_text_cleaning(item.text, clean_options) if not clean_text.strip(): continue start_sec = item.start.ordinal / 1000.0 end_sec = item.end.ordinal / 1000.0 tasks.append((i, clean_text, start_sec, end_sec)) count = 0 with concurrent.futures.ThreadPoolExecutor(max_workers=threads_count) as executor: futures = { executor.submit( process_single_line, task[0], task[1], task[2], task[3], voice_name, native_sample_rate, base_speed, pitch_steps, noise, noise_w ): task for task in tasks } for future in progress.tqdm(concurrent.futures.as_completed(futures), total=len(tasks), desc="Đang tổng hợp..."): idx, processed_chunk, start_sample, speed_applied = future.result() count += 1 if processed_chunk is not None: end_sample = start_sample + len(processed_chunk) if end_sample > len(final_audio): processed_chunk = processed_chunk[:(len(final_audio) - start_sample)] end_sample = len(final_audio) # Ghi đè âm thanh vào đúng Timeline final_audio[start_sample:end_sample] += processed_chunk yield None, log(f" [{count}/{len(subs)}] ✅ Dòng {idx+1}: Thành công | Biên độ giãn: {speed_applied:.2f}x") else: yield None, log(f" [{count}/{len(subs)}] ❌ Dòng {idx+1}: Bị bỏ qua do lỗi văn bản.") yield None, log("\n⏳ Đang áp dụng bộ lọc âm thanh (Limiter & Gain) lần cuối...") # Áp dụng âm lượng target_gain_ratio = vol_pct / 100.0 final_audio = final_audio * target_gain_ratio # Limiter mượt mà final_audio = np.tanh(final_audio) final_audio = safe_normalize(final_audio, target_peak=0.95) output_path = os.path.join(tempfile.gettempdir(), f"ketqua_Piper_Pro.wav") sf.write(output_path, final_audio, native_sample_rate) yield output_path, log(f"🎉 TỔNG HỢP HOÀN THÀNH! File xuất ra ở bên dưới.") except Exception as e: traceback.print_exc() yield None, f"❌ Lỗi hệ thống: {str(e)}" # --- 7. GIAO DIỆN GRADIO TỐI ƯU --- custom_css = """ .gradio-container {background-color: #f0faff;} #header_title {text-align: center; color: #1e40af;} .action-btn {background: linear-gradient(135deg, #60a5fa, #2563eb) !important; color: white !important; font-weight: bold;} .group-box {background: white; border-radius: 15px; padding: 20px; box-shadow: 0 4px 6px -1px rgb(0 0 0 / 0.1);} """ CLEANING_OPTIONS = [ "Bỏ qua các thẻ HTML", "Bỏ qua đoạn văn giữa dấu ngoặc đơn ()", "Bỏ qua văn bản giữa dấu xoăn {}", "Lờ chữ cái giữa hai dấu vuông []", "Bỏ qua những đoạn văn giữa dấu hoa thị *", "Bỏ qua văn bản giữa các ghi chú ♪ ♪", "Lờ đi các ký tự: * # ~" ] DEFAULT_CHECKED = [ "Bỏ qua các thẻ HTML", "Bỏ qua văn bản giữa dấu xoăn {}", "Lờ chữ cái giữa hai dấu vuông []", "Bỏ qua những đoạn văn giữa dấu hoa thị *", "Bỏ qua văn bản giữa các ghi chú ♪ ♪", "Lờ đi các ký tự: * # ~" ] user_settings = load_config() with gr.Blocks(title="Piper TTS Pro (Advanced DSP Sync)", css=custom_css) as demo: gr.Markdown("# 🎙️ Piper TTS Pro (Công Nghệ DSP & Đa Luồng Chống Sót)", elem_id="header_title") with gr.Row(): with gr.Column(scale=4): with gr.Group(elem_classes="group-box"): gr.Markdown("### 📂 1. Cấu hình Input") with gr.Row(): voice_choices = get_voice_list() default_voice = user_settings.get("voice", voice_choices[0] if voice_choices else None) v_select = gr.Dropdown(choices=voice_choices, label="Chọn Model AI (Piper)", value=default_voice, scale=4) refresh_btn = gr.Button("🔄 Tải DS", scale=1) with gr.Tabs(): with gr.TabItem("📁 Tải Lên Phụ Đề (SRT)"): srt_input = gr.File(label="Chỉ hỗ trợ file .srt chuẩn") with gr.TabItem("📝 Nhập thủ công SRT"): manual_input = gr.Textbox(label="Dán nội dung .srt vào đây", lines=6) with gr.Group(elem_classes="group-box"): gr.Markdown("### 🧹 2. Bộ Lọc Văn Bản") clean_checkboxes = gr.CheckboxGroup(choices=CLEANING_OPTIONS, value=user_settings.get("clean_opts", DEFAULT_CHECKED), label="Tùy chọn làm sạch") with gr.Column(scale=4): with gr.Group(elem_classes="group-box"): gr.Markdown("### 🎛️ 3. Cấu Hình Âm Thanh & Xử Lý") vol_pct = gr.Slider(50, 200, value=user_settings.get("vol", 150), step=10, label="Tăng cường âm lượng (%)") with gr.Row(): sp = gr.Slider(0.5, 3.0, value=user_settings.get("speed", 1.2), step=0.1, label="Tốc độ đọc mặc định") pitch = gr.Slider(-12, 12, value=user_settings.get("pitch", 0), step=1, label="Cao độ (Pitch)") # Model cục bộ ngốn CPU, khuyên dùng max 4 luồng threads_input = gr.Slider(minimum=1, maximum=10, value=user_settings.get("threads", 4), step=1, label="Số lượng luồng CPU song song") gr.Markdown("---") with gr.Row(): ns = gr.Slider(0.1, 1.0, value=user_settings.get("ns", 0.6), label="Độ nhiễu biểu cảm (Noise Scale)") nw = gr.Slider(0.1, 1.0, value=user_settings.get("nw", 0.667), label="Dao động tốc độ (Noise W)") btn = gr.Button("🚀 BẮT ĐẦU CHUYỂN ĐỔI SRT", elem_classes="action-btn", size="lg") with gr.Column(scale=4): with gr.Group(elem_classes="group-box"): gr.Markdown("### 🎧 4. Kết Quả Âm Thanh") audio_out = gr.Audio(label="File Phụ Đề Hoàn Thiện (.WAV)", type="filepath") with gr.Group(elem_classes="group-box"): gr.Markdown("### 📊 Trạng Thái Xử Lý") log_output = gr.Textbox(label="Nhật ký (Logs)", lines=12, max_lines=12, interactive=False) def refresh_voices(current): new = get_voice_list() return gr.update(choices=new, value=current if current in new else (new[0] if new else None)) refresh_btn.click(refresh_voices, [v_select], [v_select]) btn.click( process_pipeline, [v_select, srt_input, manual_input, sp, pitch, threads_input, ns, nw, vol_pct, clean_checkboxes], [audio_out, log_output] ) # Auto Save Settings settings = [v_select, clean_checkboxes, vol_pct, sp, pitch, threads_input, ns, nw] for comp in settings: comp.change(save_config_to_file, settings, None) if __name__ == "__main__": demo.queue().launch(server_name="0.0.0.0", theme=gr.themes.Soft(), css=custom_css)