import os import re import json import base64 import asyncio import requests import gradio as gr from moviepy import VideoFileClip from pydub import AudioSegment import edge_tts custom_css = """ body { background-color: #0b0f19 !important; color: #e2e8f0 !important; font-family: 'Inter', sans-serif !important; } .gradio-container { width: 100% !important; max-width: 1200px !important; margin: 0 auto !important; background: transparent !important; padding: 20px !important; box-sizing: border-box !important; } .premium-card { background: rgba(17, 24, 39, 0.7) !important; backdrop-filter: blur(16px) !important; border: 1px solid rgba(255, 255, 255, 0.08) !important; border-radius: 20px !important; padding: 25px !important; box-shadow: 0 20px 40px -15px rgba(0, 0, 0, 0.7) !important; margin-bottom: 20px !important; } .premium-btn { background: linear-gradient(135deg, #4f46e5 0%, #7c3aed 100%) !important; color: white !important; border: none !important; border-radius: 14px !important; font-weight: 700 !important; font-size: 16px !important; height: 55px !important; cursor: pointer !important; box-shadow: 0 4px 20px 0 rgba(79, 70, 229, 0.4) !important; width: 100% !important; transition: all 0.3s ease !important; } .premium-btn:hover { transform: translateY(-3px) !important; box-shadow: 0 8px 25px 0 rgba(79, 70, 229, 0.6) !important; } .premium-btn:disabled { background: #374151 !important; color: #9ca3af !important; box-shadow: none !important; cursor: not-allowed !important; transform: none !important; } @media (max-width: 768px) { .premium-card { padding: 15px !important; } } """ OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY", "") OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" def clean_khmer_text(text): """ សម្អាតអក្សរពិសេសៗដើម្បីឱ្យ Edge-TTS អានបានរលូន និងមិនគាំង """ if not text: return "" # លុបសញ្ញាដែលអាចធ្វើឱ្យ TTS គាំង ឬអានខុស text = re.sub(r'[~@#$^*()_+={}\[\]|\\<>`~]', '', text) return text.strip() def call_gemini_vision_api(optimized_video_path, total_duration): print("[INFO] STEP 3: Sending video to Gemini 2.5 Flash with strict JSON Schema...") try: with open(optimized_video_path, "rb") as f: video_base64 = base64.b64encode(f.read()).decode("utf-8") except Exception as e: raise gr.Error(f"មិនអាចអានឯកសារវីដេអូបានទេ: {str(e)}") gemini_prompt = f""" You are an elite film dubbing director. Analyze the video and audio tracks of this short movie. The total timeline of the movie is exactly {total_duration:.2f} seconds. STRICT OPERATIONAL DIRECTIVES: 1. AUDIO-FIRST GENDER CLASSIFICATION: Listen carefully to the voice pitch and tone of each spoken line. Classify the gender strictly as "Male" or "Female". 2. TIMELINE ACCURACY: Capture the exact start and end timestamps for every single dialogue dialogue utterance. 3. HIGH-QUALITY TRANSLATION: Translate the lines beautifully, naturally, and contextually into formal Khmer ("khmer_translation"). Do not literal-translate idioms. You must output a JSON object adhering exactly to this structure: {{ "total_duration": {total_duration:.2f}, "segments": [ {{ "start": 1.2, "end": 3.5, "speaker_gender": "Male", "khmer_translation": "សួស្ដី តើអ្នកសុខសប្បាយទេ?" }} ] }} """ # ប្រើប្រាស់ Structured Outputs របស់ Gemini តាមរយៈ OpenRouter ដើម្បីការពារកំហុសទម្រង់ JSON payload = { "model": "google/gemini-2.5-flash", "messages": [ { "role": "user", "content": [ {"type": "text", "text": gemini_prompt}, {"type": "image_url", "image_url": {"url": f"data:video/mp4;base64,{video_base64}"}} ] } ], "response_format": { "type": "json_object", "schema": { "type": "object", "properties": { "total_duration": {"type": "number"}, "segments": { "type": "array", "items": { "type": "object", "properties": { "start": {"type": "number"}, "end": {"type": "number"}, "speaker_gender": {"type": "string", "enum": ["Male", "Female"]}, "khmer_translation": {"type": "string"} }, "required": ["start", "end", "speaker_gender", "khmer_translation"] } } }, "required": ["total_duration", "segments"] } } } headers = { "Authorization": f"Bearer {OPENROUTER_API_KEY}", "Content-Type": "application/json" } response = requests.post(OPENROUTER_URL, json=payload, headers=headers) if response.status_code != 200: raise gr.Error(f"OpenRouter API Error (Status {response.status_code}): {response.text}") try: raw_content = response.json()['choices'][0]['message']['content'] json_clean = re.sub(r'^```json\s*|```$', '', raw_content.strip(), flags=re.IGNORECASE) return json.loads(json_clean) except Exception as e: raise gr.Error("លទ្ធផលពី Gemini មិនស្ថិតក្នុងទម្រង់ JSON ដែលត្រឹមត្រូវទេ! សូមព្យាយាមម្ដងទៀត។") async def generate_khmer_audio_segment(text, gender_type, target_duration, output_filename): clean_text = clean_khmer_text(text) if not clean_text: return None clean_gender = str(gender_type).strip().lower() voice = "km-KH-SreymomNeural" if clean_gender == "female" else "km-KH-PisethNeural" temp_mp3 = f"temp_{output_filename}.mp3" temp_wav = f"temp_raw_{output_filename}.wav" try: communicate = edge_tts.Communicate(clean_text, voice) await communicate.save(temp_mp3) except Exception as e: print(f"[ERROR] Edge-TTS failed for text: {clean_text}. Error: {e}") return None if not os.path.exists(temp_mp3): return None # បំប្លែងទៅជា WAV គុណភាពខ្ពស់ 24kHz os.system(f'ffmpeg -y -i "{temp_mp3}" -ar 24000 "{temp_wav}" > /dev/null 2>&1') if os.path.exists(temp_mp3): os.remove(temp_mp3) if not os.path.exists(temp_wav): return None audio_segment = AudioSegment.from_wav(temp_wav) current_duration = len(audio_segment) / 1000.0 # ពិនិត្យ និងកែសម្រួលល្បឿននិយាយ (Audio Speed Matching) ឱ្យត្រូវនឹង Timeline ដើម ១០០% if current_duration > target_duration and target_duration > 0: speed_factor = current_duration / target_duration # កម្រិតល្បឿនអតិបរមា 1.5x ដើម្បីកុំឱ្យសំឡេងប្រញាប់ពេកស្តាប់មិនយល់ if speed_factor > 1.5: speed_factor = 1.5 print(f"[DEBUG] Speeding up audio: {current_duration:.2f}s -> {target_duration:.2f}s ({speed_factor:.2f}x)") os.system(f'ffmpeg -y -i "{temp_wav}" -filter:a "atempo={speed_factor}" -ar 24000 "{output_filename}" > /dev/null 2>&1') else: os.system(f'ffmpeg -y -i "{temp_wav}" -ar 24000 "{output_filename}" > /dev/null 2>&1') if os.path.exists(temp_wav): os.remove(temp_wav) return output_filename async def run_vox_crown_pipeline(video_input): if not OPENROUTER_API_KEY: raise gr.Error("សូមរៀបចំដាក់ OPENROUTER_API_KEY ក្នុង Settings Secrets!") if video_input is None: raise gr.Error("សូមបញ្ចូលវីដេអូរបស់អ្នកជាមុនសិន!") video_clip = VideoFileClip(video_input) duration = video_clip.duration video_clip.close() if duration > 60.10: raise gr.Error("វីដេអូវែងហួសកំណត់ 60.10 វិនាទី!") optimized_video_path = "optimized_480p.mp4" muted_video_path = "muted_original_video.mp4" # បង្កើតវីដេអូគ្មានសំឡេង (Pure Muted Video) និងវីដេអូគំរូផ្ញើទៅ Gemini os.system(f'ffmpeg -y -i "{video_input}" -vf "scale=-2:480" -vcodec libx264 -crf 30 -preset fast -acodec aac -b:a 64k "{optimized_video_path}" > /dev/null 2>&1') os.system(f'ffmpeg -y -i "{video_input}" -an -vcodec copy "{muted_video_path}" > /dev/null 2>&1') gemini_result = call_gemini_vision_api(optimized_video_path, duration) if os.path.exists(optimized_video_path): os.remove(optimized_video_path) segments = gemini_result.get("segments", []) # បង្កើត Audio Track ទទេស្អាត (Silent) ដើម្បីកុំឱ្យមានសំឡេងរំខាន ឬ Background Music ពីវីដេអូចាស់ full_voice_track = AudioSegment.silent(duration=int(duration * 1000), frame_rate=24000) for index, seg in enumerate(segments): start_time = float(seg.get("start", 0.0)) end_time = float(seg.get("end", 0.0)) gender = seg.get("speaker_gender", "Male") text_kh = seg.get("khmer_translation", "") target_dur = end_time - start_time if target_dur <= 0 or not text_kh.strip(): continue seg_filename = f"segment_{index}.wav" await generate_khmer_audio_segment(text_kh, gender, target_dur, seg_filename) if os.path.exists(seg_filename): segment_audio = AudioSegment.from_wav(seg_filename) # Overlay ចូលទៅក្នុង Timeline ច្បាស់លាស់ full_voice_track = full_voice_track.overlay(segment_audio, position=int(start_time * 1000)) os.remove(seg_filename) final_voice_track_path = "complete_khmer_voice_track.wav" full_voice_track.export(final_voice_track_path, format="wav") final_output_movie = "vox_crown_pure_dubbed.mp4" # បញ្ចូលវីដេអូ Zin (គ្មានសំឡេងចាស់) ជាមួយ Track សំឡេងខ្មែរថ្មីសុទ្ធសាធ ១០០% ffmpeg_merge_cmd = f'ffmpeg -y -i "{muted_video_path}" -i "{final_voice_track_path}" -c:v copy -c:a aac -b:a 192k "{final_output_movie}" > /dev/null 2>&1' os.system(ffmpeg_merge_cmd) if os.path.exists(muted_video_path): os.remove(muted_video_path) if os.path.exists(final_voice_track_path): os.remove(final_voice_track_path) formatted_json_result = json.dumps(gemini_result, ensure_ascii=False, indent=2) status_msg = "🎉 [SUCCESS] ប្រព័ន្ធ VOX CROWN AI បានដំណើរការ Render វីដេអូបំប្លែងសំឡេងខ្មែរបានត្រឹមត្រូវ ១០០%!" return status_msg, final_output_movie, formatted_json_result with gr.Blocks(css=custom_css, theme=gr.themes.Default()) as demo: gr.HTML("""

VOX CROWN

""") with gr.Row(): with gr.Column(scale=5, elem_classes=["premium-card"]): gr.HTML("Media Upload") video_in = gr.Video(label="Source Movie File (Max 60.10s)", container=False) btn = gr.Button("GENERATE GEMINI DUBBING", elem_classes=["premium-btn"]) with gr.Column(scale=7, elem_classes=["premium-card"]): gr.HTML("Output Studio Player") video_out = gr.Video(label="Final Dubbed Movie Result", container=False) json_out = gr.Textbox(visible=False) status_out = gr.Textbox(visible=False) btn.click( fn=lambda: gr.update(interactive=False, value="PROCESSING DUBBING... PLEASE WAIT"), inputs=None, outputs=[btn] ).then( fn=run_vox_crown_pipeline, inputs=[video_in], outputs=[status_out, video_out, json_out] ).then( fn=lambda: gr.update(interactive=True, value="GENERATE GEMINI DUBBING"), inputs=None, outputs=[btn] ) if __name__ == "__main__": demo.launch()