import os import uuid import threading import asyncio import requests import json import time import subprocess import logging import numpy as np from flask import Flask, request, jsonify, render_template_string, send_from_directory import whisper import edge_tts # --- KONFIGURASI SILENT LOGS --- os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' logging.getLogger('werkzeug').setLevel(logging.ERROR) app = Flask(__name__) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) UPLOAD_FOLDER = os.path.join(BASE_DIR, 'uploads') app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER os.makedirs(UPLOAD_FOLDER, exist_ok=True) tasks = {} # --- MAP SUARA (MALE & FEMALE) --- VOICE_MAP = { 'id-ID': {'Male': 'id-ID-ArdiNeural', 'Female': 'id-ID-GadisNeural'}, 'en-US': {'Male': 'en-US-ChristopherNeural', 'Female': 'en-US-AriaNeural'}, 'ja-JP': {'Male': 'ja-JP-KeitaNeural', 'Female': 'ja-JP-NanamiNeural'} } # Mapping Bahasa untuk Prompt AI LANG_MAP = { 'id-ID': 'Indonesia', 'en-US': 'Inggris', 'ja-JP': 'Jepang' } # Load Whisper (CPU Friendly, FP16 Fixed) whisper_model = whisper.load_model("base") def get_audio_duration(file_path): cmd = [ 'ffprobe', '-v', 'error', '-show_entries', 'format=duration', '-of', 'default=noprint_wrappers=1:nokey=1', file_path ] result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) try: return float(result.stdout) except: return 0.0 def analyze_gender_and_pitch(audio_path): """Menganalisis potongan audio untuk menentukan gender dan variasi pitch.""" try: import librosa # Load audio dengan sample rate standard y, sr = librosa.load(audio_path, sr=22050) if len(y) == 0: return "Male", "+0Hz" # Deteksi Fundamental Frequency (F0) f0 = librosa.yin(y, fmin=65, fmax=300) valid_f0 = f0[~np.isnan(f0)] if len(valid_f0) > 0: mean_f0 = np.mean(valid_f0) # Threshold umum: > 165Hz = Perempuan, < 165Hz = Laki-laki gender = "Female" if mean_f0 >= 165 else "Male" # Hitung variasi pitch (agar tiap orang suaranya beda) # Normal cowok ~120Hz, cewek ~210Hz. Dibagi 2 agar tidak terlalu ekstrem base_f0 = 210.0 if gender == "Female" else 120.0 pitch_shift = int((mean_f0 - base_f0) / 2) # Batasi modifikasi pitch Edge TTS agar tidak rusak (antara -20Hz s/d +20Hz) pitch_shift = max(-20, min(20, pitch_shift)) pitch_str = f"+{pitch_shift}Hz" if pitch_shift >= 0 else f"{pitch_shift}Hz" return gender, pitch_str except Exception as e: print(f"Pitch analysis warning: {e}") return "Male", "+0Hz" # Default fallback def translate_segments_llm(segments, custom_prompt, target_voice): target_lang = LANG_MAP.get(target_voice, 'Indonesia') # PERBAIKAN: Memasukkan bahasa target secara paksa ke dalam prompt if custom_prompt: instruction = f"{custom_prompt}\n\nPENTING: Terjemahkan SEMUA teks ke dalam bahasa {target_lang}." else: instruction = f"Terjemahkan teks dalam JSON ini ke bahasa {target_lang} dengan akurat. Balas HANYA dengan JSON array." input_data = [{"id": i, "text": s['text']} for i, s in enumerate(segments)] full_prompt = f"{instruction}\n\nFormat: [{{'id': 0, 'text': '...'}}]\n\nData:\n{json.dumps(input_data)}" url = "https://www.puruboy.kozow.com/api/ai/notegpt" payload = {"prompt": full_prompt, "model": "gemini-3-flash-preview", "chat_mode": "standard"} try: response = requests.post(url, json=payload, timeout=60) full_text = "" for line in response.iter_lines(): if line: decoded = line.decode('utf-8') if decoded.startswith("data: "): data = json.loads(decoded[6:]) full_text += data.get("text", "") start_idx = full_text.find('[') end_idx = full_text.rfind(']') + 1 translated_list = json.loads(full_text[start_idx:end_idx]) for item in translated_list: segments[item['id']]['translated_text'] = item['text'] except Exception as e: print(f"Translation Error: {e}") for s in segments: s['translated_text'] = s['text'] return segments # PERBAIKAN: Menambahkan parameter pitch async def generate_tts(text, voice, path, pitch_str="+0Hz"): communicate = edge_tts.Communicate(text, voice, pitch=pitch_str) await communicate.save(path) def process_dubbing(task_id, video_path, target_voice, custom_prompt): try: tasks[task_id]['status'] = 'Mengekstrak Audio...' orig_audio = os.path.join(app.config['UPLOAD_FOLDER'], f"{task_id}_orig.wav") subprocess.run(['ffmpeg', '-loglevel', 'quiet', '-y', '-i', video_path, '-vn', '-acodec', 'pcm_s16le', '-ar', '44100', orig_audio], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) tasks[task_id]['status'] = 'Transkripsi...' result = whisper_model.transcribe(orig_audio, verbose=False, fp16=False) segments = result['segments'] tasks[task_id]['status'] = f'Translasi AI ({LANG_MAP.get(target_voice, target_voice)})...' # Pass target_voice ke translator translated_segments = translate_segments_llm(segments, custom_prompt, target_voice) tasks[task_id]['status'] = 'Menganalisis Suara & Dubbing...' processed_audio_files = [] for i, seg in enumerate(translated_segments): start_t = seg['start'] end_t = seg['end'] duration_orig = end_t - start_t text = seg.get('translated_text', seg['text']) if not text.strip(): continue # Potong audio asli khusus untuk segmen ini guna deteksi suara chunk_wav = os.path.join(app.config['UPLOAD_FOLDER'], f"{task_id}_chunk_{i}.wav") subprocess.run(['ffmpeg', '-loglevel', 'quiet', '-y', '-i', orig_audio, '-ss', str(start_t), '-t', str(duration_orig), chunk_wav], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # Deteksi Cewek/Cowok dan variasi pitch gender, pitch_str = analyze_gender_and_pitch(chunk_wav) # Pilih Voice ID yang sesuai berdasarkan bahasa dan gender selected_voice = VOICE_MAP.get(target_voice, VOICE_MAP['id-ID'])[gender] raw_tts = os.path.join(app.config['UPLOAD_FOLDER'], f"{task_id}_raw_{i}.mp3") sync_tts = os.path.join(app.config['UPLOAD_FOLDER'], f"{task_id}_sync_{i}.wav") # Generate TTS dengan pitch modifier asyncio.run(generate_tts(text, selected_voice, raw_tts, pitch_str)) tts_dur = get_audio_duration(raw_tts) speed = min(max(tts_dur / duration_orig, 0.7), 1.8) if duration_orig > 0 else 1.0 subprocess.run(['ffmpeg', '-loglevel', 'quiet', '-y', '-i', raw_tts, '-filter:a', f'atempo={speed}', '-ar', '44100', sync_tts], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) processed_audio_files.append({'path': sync_tts, 'start': start_t}) tasks[task_id]['status'] = 'Mixing Audio & Rendering...' output_filename = f"{task_id}_output.mp4" output_path = os.path.join(app.config['UPLOAD_FOLDER'], output_filename) # LOGIKA AUDIO BARU: filter_complex = "[0:a]equalizer=f=1000:width_type=o:w=2:g=-15,volume=0.4[bg];" inputs_cmd = ['ffmpeg', '-loglevel', 'quiet', '-y', '-i', video_path] amix_inputs = "[bg]" for i, item in enumerate(processed_audio_files): idx = i + 1 inputs_cmd.extend(['-i', item['path']]) start_ms = int(item['start'] * 1000) filter_complex += f"[{idx}:a]adelay={start_ms}|{start_ms},volume=3.0[dub{idx}];" amix_inputs += f"[dub{idx}]" filter_complex += f"{amix_inputs}amix=inputs={len(processed_audio_files)+1}:duration=first:dropout_transition=0,volume=1.5[outa]" final_cmd = inputs_cmd + [ '-filter_complex', filter_complex, '-map', '0:v', '-map', '[outa]', '-c:v', 'libx264', '-preset', 'ultrafast', '-c:a', 'aac', '-b:a', '192k', output_path ] subprocess.run(final_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # Cleanup file temporary for file in os.listdir(app.config['UPLOAD_FOLDER']): if task_id in file and not file.endswith("_output.mp4"): try: os.remove(os.path.join(app.config['UPLOAD_FOLDER'], file)) except: pass tasks[task_id]['status'] = 'Selesai' tasks[task_id]['result_video'] = f"/download/{output_filename}" except Exception as e: tasks[task_id]['status'] = 'Error' tasks[task_id]['error_message'] = str(e) # --- ROUTES --- @app.route('/') def index(): return render_template_string(HTML_TEMPLATE) @app.route('/generate', methods=['POST']) def generate(): file = request.files.get('video') if not file: return jsonify({'error': 'No file'}) task_id = str(uuid.uuid4()) path = os.path.join(app.config['UPLOAD_FOLDER'], f"{task_id}.mp4") file.save(path) tasks[task_id] = {'status': 'Queued', 'result_video': None, 'error_message': None} threading.Thread(target=process_dubbing, args=(task_id, path, request.form.get('voice'), request.form.get('prompt'))).start() return jsonify({'task_id': task_id}) @app.route('/status') def status(): return jsonify(tasks.get(request.args.get('task_id'), {})) @app.route('/download/') def download(f): return send_from_directory(app.config['UPLOAD_FOLDER'], f) # --- HTML DENGAN TAILWIND CSS --- HTML_TEMPLATE = """ AI Dubbing Pro

🎙️ Dubbing Sync Pro

Deteksi Gender & Multi-Speaker Auto-Pitch

""" if __name__ == '__main__': app.run(host='0.0.0.0', port=7860)