app.py

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()