import os import requests import json import time import subprocess import gradio as gr import uuid import asyncio import edge_tts from dotenv import load_dotenv # Load environment variables load_dotenv() # API Key B_KEY = os.getenv("B_KEY") # URL API_URL = os.getenv("API_URL") UPLOAD_URL = os.getenv("UPLOAD_URL") async def get_voices(): voices = await edge_tts.list_voices() return [(voice['ShortName'], voice['ShortName']) for voice in voices] async def text_to_speech(voice_id, text, session_id): communicate = edge_tts.Communicate(text, voice_id) audio_file_path = f'temp_voice_{session_id}.mp3' await communicate.save(audio_file_path) return audio_file_path def upload_file(file_path): with open(file_path, 'rb') as file: files = {'fileToUpload': (os.path.basename(file_path), file)} data = {'reqtype': 'fileupload'} response = requests.post(UPLOAD_URL, files=files, data=data) if response.status_code == 200: return response.text.strip() return None def lipsync_api_call(video_url, audio_url): headers = { "Content-Type": "application/json", "x-api-key": B_KEY } data = { "audioUrl": audio_url, "videoUrl": video_url, "maxCredits": 1000, "model": "sync-1.6.0", "synergize": True, "pads": [0, 5, 0, 0], "synergizerStrength": 1 } response = requests.post(API_URL, headers=headers, data=json.dumps(data)) return response.json() def check_job_status(job_id): headers = {"x-api-key": B_KEY} max_attempts = 30 # Limit the number of attempts for _ in range(max_attempts): response = requests.get(f"{API_URL}/{job_id}", headers=headers) data = response.json() if data["status"] == "COMPLETED": return data["videoUrl"] elif data["status"] == "FAILED": return None time.sleep(10) return None def get_media_duration(file_path): # Fetch media duration using ffprobe 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.PIPE) return float(result.stdout.strip()) def combine_audio_video(video_path, audio_path, output_path): # Get durations of both video and audio video_duration = get_media_duration(video_path) audio_duration = get_media_duration(audio_path) if video_duration > audio_duration: # Trim video to match the audio length cmd = [ 'ffmpeg', '-i', video_path, '-i', audio_path, '-t', str(audio_duration), # Trim video to audio duration '-map', '0:v', '-map', '1:a', '-c:v', 'copy', '-c:a', 'aac', '-y', output_path ] else: # Loop video if it's shorter than audio loop_count = int(audio_duration // video_duration) + 1 # Calculate how many times to loop cmd = [ 'ffmpeg', '-stream_loop', str(loop_count), '-i', video_path, '-i', audio_path, '-t', str(audio_duration), # Match the duration of the final video with the audio '-map', '0:v', '-map', '1:a', '-c:v', 'copy', '-c:a', 'aac', '-shortest', '-y', output_path ] subprocess.run(cmd, check=True) async def process_video(voice, video_url, text, progress=gr.Progress()): session_id = str(uuid.uuid4()) # Generate a unique session ID progress(0, desc="Generating speech...") audio_path = await text_to_speech(voice, text, session_id) if not audio_path: return None, "Failed to generate speech audio." progress(0.2, desc="Processing video...") try: progress(0.3, desc="Uploading audio...") audio_url = upload_file(audio_path) if not audio_url: raise Exception("Failed to upload audio file") progress(0.4, desc="Initiating lipsync...") job_data = lipsync_api_call(video_url, audio_url) if "error" in job_data or "message" in job_data: raise Exception(job_data.get("error", job_data.get("message", "Unknown error"))) job_id = job_data["id"] progress(0.5, desc="Processing lipsync...") result_url = check_job_status(job_id) if result_url: progress(0.9, desc="Downloading result...") response = requests.get(result_url) output_path = f"output_{session_id}.mp4" with open(output_path, "wb") as f: f.write(response.content) progress(1.0, desc="Complete!") return output_path, "Lipsync completed successfully!" else: raise Exception("Lipsync processing failed or timed out") except Exception as e: progress(0.8, desc="Falling back to simple combination...") try: # Download the video from the URL video_response = requests.get(video_url) video_path = f"temp_video_{session_id}.mp4" with open(video_path, "wb") as f: f.write(video_response.content) output_path = f"output_{session_id}.mp4" combine_audio_video(video_path, audio_path, output_path) progress(1.0, desc="Complete!") return output_path, f"Used fallback method. Original error: {str(e)}" except Exception as fallback_error: return None, f"All methods failed. Error: {str(fallback_error)}" finally: # Cleanup if os.path.exists(audio_path): os.remove(audio_path) if os.path.exists(f"temp_video_{session_id}.mp4"): os.remove(f"temp_video_{session_id}.mp4") def create_interface(): async def populate_voices(): voices = await get_voices() return [v[1] for v in voices] with gr.Blocks() as app: gr.Markdown("# JSON Train") with gr.Row(): with gr.Column(): voice_dropdown = gr.Dropdown(label="Select Voice", choices=asyncio.run(populate_voices())) video_url_input = gr.Textbox(label="Enter Video URL") text_input = gr.Textbox(label="Enter text", lines=3) generate_btn = gr.Button("Generate Video") with gr.Column(): video_output = gr.Video(label="Generated Video") status_output = gr.Textbox(label="Status", interactive=False) async def on_generate(voice_name, video_url, text): return await process_video(voice_name, video_url, text) generate_btn.click( fn=on_generate, inputs=[voice_dropdown, video_url_input, text_input], outputs=[video_output, status_output] ) return app if __name__ == "__main__": app = create_interface() app.launch()