import os import time import random import base64 import tempfile import requests import numpy as np import soundfile as sf import pysrt import librosa import pyrubberband as pyrb import concurrent.futures import gradio as gr # --- MAP ID GIỌNG ĐỌC TIKTOK --- voice_id_mapping = { "🇻🇳 Nữ (Việt Nam) - BV074": "BV074_streaming", "🇻🇳 Nam (Việt Nam) - BV075": "BV075_streaming", "🇺🇸 Nữ US (Jessie)": "en_us_002", "🇺🇸 Nam US (Joey)": "en_us_006", "🇺🇸 Nam kể chuyện (Story Teller)": "en_male_narration", "🎭 Ghost Face (Kinh dị)": "en_us_ghostface", "🎶 Hát hài hước": "en_male_sing_funny_it_goes_up" } # --- 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) if np.max(np.abs(audio_array)) > 1.0: audio_array = audio_array / 32768.0 audio_array = audio_array - np.mean(audio_array) 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 # --- HÀM GỌI API CÓ TÍCH HỢP CHỐNG SPAM VÀ THỬ LẠI --- def get_tiktok_audio_with_retry(text, v_code, target_sr=None, max_retries=5): url = "https://tiktok-tts.weilnet.workers.dev/api/generation" payload = {"text": text, "voice": v_code} for attempt in range(max_retries): time.sleep(random.uniform(0.1, 0.8)) try: response = requests.post(url, json=payload, timeout=20) if response.status_code in [429, 500, 502, 503, 504]: wait_time = (2 ** attempt) + random.uniform(0.5, 1.5) time.sleep(wait_time) continue response.raise_for_status() data = response.json() if "data" in data: audio_bytes = base64.b64decode(data["data"]) with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f: f.write(audio_bytes) tmp_path = f.name audio_array, sr = librosa.load(tmp_path, sr=target_sr) os.remove(tmp_path) return audio_array, sr else: return None, None except requests.exceptions.RequestException: wait_time = (2 ** attempt) + random.uniform(0.5, 1.5) time.sleep(wait_time) return None, None def process_single_line(index, text, start_sec, end_sec, v_code, native_sr, speed_m, pitch_steps): srt_duration = end_sec - start_sec if srt_duration <= 0: srt_duration = 0.5 audio_chunk, _ = get_tiktok_audio_with_retry(text, v_code, target_sr=native_sr) if audio_chunk is None: return index, None, None, None audio_chunk = safe_normalize(audio_chunk, target_peak=0.90) orig_dur = len(audio_chunk) / native_sr required_speed = orig_dur / srt_duration speed = max(speed_m, required_speed) speed = min(speed, 2.0) # Ép tốc độ processed_chunk = pyrb.time_stretch(audio_chunk, native_sr, speed) # Ép cao độ (Pitch Shift) if pitch_steps != 0: processed_chunk = pyrb.pitch_shift(processed_chunk, native_sr, pitch_steps) 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, speed # --- HÀM XỬ LÝ CHÍNH KẾT NỐI VỚI GRADIO --- def srt_to_speech_handler(srt_file, chon_giong, toc_do, cao_do, so_luong_luong, progress=gr.Progress(track_tqdm=True)): if srt_file is None: return None, "❌ Vui lòng tải lên file phụ đề .srt trước!" log_messages = [] def log(msg): log_messages.append(msg) return "\n".join(log_messages) voice_id = voice_id_mapping.get(chon_giong) yield None, log("⏳ Đang kiểm tra kết nối và trích xuất tần số góc từ TikTok...") _, native_sample_rate = get_tiktok_audio_with_retry("Test", voice_id, max_retries=2) if native_sample_rate is None: yield None, log("❌ Không thể kết nối tới API của TikTok. Vui lòng thử lại sau.") return yield None, log(f"✅ Kết nối thành công! Tần số gốc: {native_sample_rate} Hz\n⏳ Đang đọc file phụ đề...") try: subs = pysrt.open(srt_file.name) except Exception as e: yield None, log(f"❌ Lỗi đọc file SRT: {str(e)}") return if len(subs) == 0: yield None, log("❌ File SRT trống, không tìm thấy câu phụ đề nào.") return max_end_sec = max([sub.end.ordinal / 1000.0 for sub in subs]) total_samples = int(max_end_sec * native_sample_rate) final_audio = np.zeros(total_samples, dtype=np.float32) yield None, log(f"🚀 Bắt đầu tổng hợp đa luồng ({so_luong_luong} luồng) cho {len(subs)} câu...") tasks = [] for i, sub in enumerate(subs): text = sub.text.replace('\n', ' ').strip() if not text: continue if len(text) > 290: text = text[:290] start_sec = sub.start.ordinal / 1000.0 end_sec = sub.end.ordinal / 1000.0 tasks.append((i, text, start_sec, end_sec)) count = 0 with concurrent.futures.ThreadPoolExecutor(max_workers=so_luong_luong) as executor: futures = { executor.submit( process_single_line, task[0], task[1], task[2], task[3], voice_id, native_sample_rate, toc_do, cao_do ): task for task in tasks } for future in progress.tqdm(concurrent.futures.as_completed(futures), total=len(tasks), desc="Đang xử lý phụ đề"): 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) final_audio[start_sample:end_sample] += processed_chunk yield None, log(f" [{count}/{len(subs)}] ✅ Dòng {idx+1}: Thành công | Tốc độ: {speed_applied:.2f}x") else: yield None, log(f" [{count}/{len(subs)}] ❌ Dòng {idx+1}: Thất bại hoàn toàn.") yield None, log("\n⏳ Đang áp dụng bộ lọc âm thanh Limiter lần cuối...") 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_TikTok_Smart.wav") sf.write(output_path, final_audio, native_sample_rate) yield output_path, log("🎉 TỔNG HỢP HOÀN THÀNH! Bạn có thể tải file ở bên dưới.") # --- KHỞI TẠO GIAO DIỆN GRADIO (ĐÃ FIX TƯƠNG THÍCH GRADIO 6.0) --- with gr.Blocks(title="TikTok TTS Pro") as demo: gr.Markdown("

🎙️ Chuyển Đổi Phụ Đề (SRT) Sang Giọng Đọc TikTok

") gr.Markdown("

Hệ thống tự động ép tốc độ, chỉnh cao độ và xử lý đa luồng chống sót câu.

") with gr.Row(): # CỘT TRÁI: Nhập liệu và Cài đặt with gr.Column(scale=5): with gr.Group(): gr.Markdown("### 📂 1. Tải Lên Phụ Đề") srt_input = gr.File(label="Tải lên duy nhất 1 file (.srt)", file_types=[".srt"]) with gr.Group(): gr.Markdown("### 🎛️ 2. Cấu Hình Giọng Đọc") voice_input = gr.Dropdown( choices=list(voice_id_mapping.keys()), value="🇻🇳 Nữ (Việt Nam) - BV074", label="Chọn Giọng", interactive=True ) with gr.Row(): speed_input = gr.Slider(minimum=1.0, maximum=1.5, value=1.2, step=0.1, label="Tốc độ mặc định") pitch_input = gr.Slider(minimum=-12, maximum=12, value=0, step=1, label="Cao độ (Pitch)") threads_input = gr.Slider(minimum=1, maximum=20, value=10, step=1, label="Số lượng luồng tải song song") btn_run = gr.Button("🚀 BẮT ĐẦU CHUYỂN ĐỔI", variant="primary", size="lg") # CỘT PHẢI: Kết quả đầu ra và Logs with gr.Column(scale=5): with gr.Group(): gr.Markdown("### 🎧 3. Kết Quả Âm Thanh") audio_output = gr.Audio(label="File Phụ Đề Hoàn Thiện (.WAV)", type="filepath") with gr.Group(): gr.Markdown("### 📊 Trạng Thái Xử Lý") log_output = gr.Textbox(label="Nhật ký (Logs)", lines=14, max_lines=14, interactive=False) # Gắn sự kiện nút bấm btn_run.click( fn=srt_to_speech_handler, inputs=[srt_input, voice_input, speed_input, pitch_input, threads_input], outputs=[audio_output, log_output] ) if __name__ == "__main__": # Đã cấu hình theme tại hàm launch theo chuẩn Gradio 6.0 demo.queue().launch(theme=gr.themes.Soft())