import os import sys import subprocess import tempfile from pathlib import Path from dotenv import load_dotenv import whisper import gradio as gr import azure.cognitiveservices.speech as speechsdk import requests from pydub import AudioSegment from pydub.utils import make_chunks import shutil import io import asyncio import json # Limit OMP threads (fix libgomp issue) os.environ["OMP_NUM_THREADS"] = os.getenv("OMP_NUM_THREADS", "1") # Only load .env if running locally (not in Spaces) if os.getenv("SPACE_ID") is None: load_dotenv() AZURE_TRANSLATOR_KEY = os.getenv("AZURE_TRANSLATOR_KEY", "").strip() AZURE_TRANSLATOR_REGION = os.getenv("AZURE_TRANSLATOR_REGION", "").strip() AZURE_KEY = os.getenv("AZURE_KEY", "").strip() AZURE_REGION = os.getenv("AZURE_REGION", "").strip() # Validate API keys missing = [] if not AZURE_TRANSLATOR_KEY: missing.append("AZURE_TRANSLATOR_KEY") if not AZURE_TRANSLATOR_REGION: missing.append("AZURE_TRANSLATOR_REGION") if not AZURE_KEY: missing.append("AZURE_KEY") if not AZURE_REGION: missing.append("AZURE_REGION") if missing: sys.exit(f"❌ Missing environment variables: {', '.join(missing)}") # --- Language map --- LANGUAGE_MAP = { "French": "fr", "German": "de", "Italian": "it", "Japanese": "ja", "Dutch": "nl", "Swedish": "sv", "Spanish": "es", "Polish": "pl", "Arabic": "ar", } # --- Helper function for SRT formatting --- def _format_time(seconds): """Converts seconds to SRT time format (HH:MM:SS,mmm).""" h = int(seconds / 3600) m = int((seconds % 3600) / 60) s = int(seconds % 60) ms = int((seconds - int(seconds)) * 1000) return f"{h:02}:{m:02}:{s:02},{ms:03}" # --- Async TTS helper function --- async def _synthesize_tts_async(speech_config, text): loop = asyncio.get_running_loop() synthesizer = speechsdk.SpeechSynthesizer(speech_config=speech_config, audio_config=None) # Run blocking .get() in thread executor result = await loop.run_in_executor( None, lambda: synthesizer.speak_text_async(text).get() ) if result.reason != speechsdk.ResultReason.SynthesizingAudioCompleted: print(f"TTS synthesis failed with reason: {result.reason}") return AudioSegment.silent(duration=1000) # Return silent audio as fallback audio_data = result.audio_data if not audio_data: print("No audio data received from TTS") return AudioSegment.silent(duration=1000) try: # Save to temp file for pydub processing with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_wav: temp_wav.write(audio_data) temp_wav_path = temp_wav.name # Load with pydub audio_segment = AudioSegment.from_wav(temp_wav_path) # Clean up temp file os.unlink(temp_wav_path) return audio_segment except Exception as e: print(f"Error processing TTS audio: {e}") # Fallback: try to create silent audio of estimated length try: estimated_duration = len(text.split()) * 0.3 # Rough estimate: 0.3s per word return AudioSegment.silent(duration=int(estimated_duration * 1000)) except: return AudioSegment.silent(duration=1000) # --- Main dubbing function --- async def dub_video(uploaded_video_path, target_lang_name, voice_gender): print(f"Received inputs: video={uploaded_video_path}, lang={target_lang_name}, voice={voice_gender}") target_lang_code = LANGUAGE_MAP.get(target_lang_name) if not target_lang_code: return None, None, "❌ Error: Invalid language selected." with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) video_in = temp_path / "input_video.mp4" audio_wav = temp_path / "input.wav" dubbed_audio_path = temp_path / "dubbed.wav" output_video_temp = temp_path / "output_dubbed.mp4" output_subtitles_temp = temp_path / "subtitles.srt" shutil.copy(uploaded_video_path, video_in) print("🎧 Extracting audio...") subprocess.run(["ffmpeg", "-y", "-i", video_in, "-ac", "1", "-ar", "16000", audio_wav]) print("📝 Transcribing (Whisper)...") model = whisper.load_model("large") result = model.transcribe(str(audio_wav), language="en") segments = result["segments"] print(f"🌐 Translating to {target_lang_name}...") english_lines = [seg["text"].strip() for seg in segments] translated_lines = [] endpoint = f"https://{AZURE_TRANSLATOR_REGION}.api.cognitive.microsoft.com" headers = { "Ocp-Apim-Subscription-Key": AZURE_TRANSLATOR_KEY, "Ocp-Apim-Subscription-Region": AZURE_TRANSLATOR_REGION, "Content-Type": "application/json", "Accept": "application/json" } for line in english_lines: if line: # Only translate non-empty lines body = [{"text": line}] response = requests.post( f"{endpoint}/translator/text/v3.0/translate?api-version=3.0&from=en&to={target_lang_code}", headers=headers, json=body ) if response.status_code == 200: translations = response.json() translated_text = translations[0]["translations"][0]["text"] translated_lines.append(translated_text) else: print(f"Translation error: {response.status_code} - {response.text}") translated_lines.append(line) # Fallback to original else: translated_lines.append("") print(f"Translated lines:\n{translated_lines}") # --- LANGUAGE DETECTION + AUTO-CORRECTION --- from langdetect import detect for i, line in enumerate(translated_lines): if line: # Skip empty lines detected_lang = detect(line) if detected_lang != target_lang_code: print(f"⚠️ Warning: Detected {detected_lang}, correcting to {target_lang_code}...") try: body = [{"text": line}] response = requests.post( f"{endpoint}/translator/text/v3.0/translate?api-version=3.0&from={detected_lang}&to={target_lang_code}", headers=headers, json=body ) if response.status_code == 200: translations = response.json() corrected_text = translations[0]["translations"][0]["text"] translated_lines[i] = corrected_text print(f"✅ Corrected: {corrected_text}") else: print(f"❌ Correction failed ({response.status_code}) - keeping original line.") except Exception as e: print(f"❌ Error correcting translation: {e}") print("🔊 Generating speech with Azure Neural TTS...") voice_map = { "fr": {"female": "fr-FR-DeniseNeural", "male": "fr-FR-HenriNeural"}, "de": {"female": "de-DE-KatjaNeural", "male": "de-DE-ConradNeural"}, "it": {"female": "it-IT-ElsaNeural", "male": "it-IT-DiegoNeural"}, "ja": {"female": "ja-JP-NanamiNeural", "male": "ja-JP-KeitaNeural"}, "nl": {"female": "nl-NL-ColetteNeural", "male": "nl-NL-MaartenNeural"}, "sv": {"female": "sv-SE-HilleviNeural", "male": "sv-SE-MattiasNeural"}, "es": {"female": "es-ES-ElviraNeural", "male": "es-ES-AlvaroNeural"}, "pl": {"female": "pl-PL-AgnieszkaNeural", "male": "pl-PL-MarekNeural"}, "ar": {"female": "ar-SA-ZariyahNeural", "male": "ar-SA-HamedNeural"}, } selected_voice = voice_map.get(target_lang_code, {}).get(voice_gender) if not selected_voice: return None, None, f"❌ Error: Voice for {target_lang_name} ({voice_gender}) not found." print(f"Using TTS voice: {selected_voice}") speech_config = speechsdk.SpeechConfig(subscription=AZURE_KEY, region=AZURE_REGION) speech_config.speech_synthesis_voice_name = selected_voice tasks = [_synthesize_tts_async(speech_config, translated_text) for translated_text in translated_lines] segment_audios = await asyncio.gather(*tasks) full_audio = AudioSegment.silent(duration=segments[-1]["end"] * 1000) for seg, segment_audio in zip(segments, segment_audios): start_ms = int(seg["start"] * 1000) full_audio = full_audio.overlay(segment_audio, position=start_ms) print("🎥 Merging dubbed audio into video...") full_audio.export(str(dubbed_audio_path), format="wav") subprocess.run([ "ffmpeg", "-y", "-i", str(video_in), "-i", str(dubbed_audio_path), "-c:v", "copy", "-map", "0:v:0", "-map", "1:a:0", "-map", "-0:a", str(output_video_temp) ]) print("📄 Generating subtitle file...") srt_content = "" for i, (seg, translated_text) in enumerate(zip(segments, translated_lines)): start_time = _format_time(seg["start"]) end_time = _format_time(seg["end"]) srt_content += f"{i + 1}\n" srt_content += f"{start_time} --> {end_time}\n" srt_content += f"{translated_text}\n\n" output_subtitles_temp.write_text(srt_content, encoding="utf-8") print("✅ Done!") output_dir = Path(tempfile.mkdtemp(prefix="dubbed_output_")) output_video_path = output_dir / "output_dubbed.mp4" output_subtitles_path = output_dir / "subtitles.srt" shutil.copy(output_video_temp, output_video_path) shutil.copy(output_subtitles_temp, output_subtitles_path) return str(output_video_path), str(output_subtitles_path), "✅ Done! Your video and subtitles are ready." # --- Gradio UI setup --- with gr.Blocks(title="AI Video Dubber") as demo: gr.Markdown("## 🎬 AI Video Dubber App") gr.Markdown("Upload an English video, choose a language and voice, and get a synced dubbed version.") with gr.Row(): with gr.Column(): uploaded_video = gr.Video(label="📤 Upload your video") target_lang_choices = list(LANGUAGE_MAP.keys()) target_lang_dropdown = gr.Dropdown( label="🌍 Target language", choices=target_lang_choices, value=target_lang_choices[0], ) voice_gender_dropdown = gr.Dropdown( label="🎙️ Voice Gender", choices=["female", "male"], value="female" ) run_button = gr.Button("🚀 Start Dubbing") with gr.Column(): dubbed_video_out = gr.Video(label="Dubbed Video") download_subtitles = gr.File(label="Download Subtitle File") status_message = gr.Textbox(label="Status") run_button.click( fn=dub_video, inputs=[uploaded_video, target_lang_dropdown, voice_gender_dropdown], outputs=[dubbed_video_out, download_subtitles, status_message] ) if __name__ == "__main__": demo.launch()