app.py

import tempfile
import os
import shutil
import librosa
import json
import subprocess
import gc
import requests
import time
import random
import re
from googletrans import Translator
import asyncio
from flask import Flask, request, jsonify, send_from_directory
from omegaconf import OmegaConf
import torch
from diffusers import AutoencoderKL, DDIMScheduler
from latentsync.models.unet import UNet3DConditionModel
from latentsync.pipelines.lipsync_pipeline import LipsyncPipeline
from diffusers.utils.import_utils import is_xformers_available
from accelerate.utils import set_seed
from latentsync.whisper.audio2feature import Audio2Feature
from openai import OpenAI
from elevenlabs import set_api_key, generate, play, clone, Voice, VoiceSettings
from torch.cuda.amp import autocast
 
# Initialize the Flask app
app = Flask(__name__)
TEMP_DIR = None
VIDEO_DIRECTORY = os.path.abspath("videos")
os.makedirs(VIDEO_DIRECTORY, exist_ok=True)

# HeyGen API Configuration
HEYGEN_API_KEY = "NGM2N2VjNmM4NWM0NGQxMjkyNWFiMjg4OTdlMTI2MDItMTcyNDQ5ODM1MA=="
HEYGEN_GENERATE_URL = "https://api.heygen.com/v2/video/generate"
HEYGEN_STATUS_URL = "https://api.heygen.com/v1/video_status.get"

# Initialize OpenAI client
client = OpenAI(api_key="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")

def clear_cuda_memory():
    torch.cuda.empty_cache()
    gc.collect()

def openai_chat_avatar(text_prompt):
    """Summarize text using OpenAI GPT-4o-mini"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Summarize the following paragraph into a complete and accurate single sentence with no more than 30 words. The summary should capture the gist of the paragraph and make sense and remove the citation and document name from the end."},
            {"role": "user", "content": f"Please summarize the following paragraph into one sentence with 30 words or fewer, ensuring it makes sense and captures the gist and remove the citation from the end: {text_prompt}"},
        ],
        max_tokens = len(text_prompt),
    )
    return response

def ryzedb_chat_avatar(question, app_id):
    """Query RyzeDB API for response"""
    url = "https://inference.dev.ryzeai.ai/v2/chat/stream"
    print("ryze db question", question)
    
    payload = {
        "input": {
            "app_id": app_id,
            "query": question,
            "chat_history": []
        },
        "config": {
            "thread_id": "123456"
        }
    }
    
    headers = {
        'Content-Type': 'application/json'
    }
    
    response = requests.post(url, json=payload, headers=headers, stream=True)
    
    try:
        raw_text = response.text.strip()
    
        if raw_text.startswith("data:"):
            raw_text = raw_text[len("data:"):].strip()
    
        json_data = json.loads(raw_text)
    
        response_content = json_data.get("content", "")
        return response_content

    except Exception as e:
        print("Error parsing response:", e)
        return ""

def run_inference(video_path, audio_path, video_out_path,
                  inference_ckpt_path, unet_config_path="configs/unet/second_stage.yaml",
                  inference_steps=20, guidance_scale=1.0, seed=1247):
    clear_cuda_memory()
    
    # Load configuration
    config = OmegaConf.load(unet_config_path)
 
    # Determine proper dtype based on GPU capabilities
    is_fp16_supported = torch.cuda.is_available() and torch.cuda.get_device_capability()[0] > 7
    dtype = torch.float16 if is_fp16_supported else torch.float32
 
    # Setup scheduler
    scheduler = DDIMScheduler.from_pretrained("configs")
 
    # Choose whisper model based on config settings
    if config.model.cross_attention_dim == 768:
        whisper_model_path = "checkpoints/whisper/small.pt"
    elif config.model.cross_attention_dim == 384:
        whisper_model_path = "checkpoints/whisper/tiny.pt"
    else:
        raise NotImplementedError("cross_attention_dim must be 768 or 384")
 
    # Initialize the audio encoder
    audio_encoder = Audio2Feature(model_path=whisper_model_path,
                                  device="cuda", num_frames=config.data.num_frames)
 
    # Load VAE
    vae = AutoencoderKL.from_pretrained("stabilityai/sd-vae-ft-mse", torch_dtype=dtype)
    vae.config.scaling_factor = 0.18215
    vae.config.shift_factor = 0
 
    # Load UNet model from the checkpoint
    unet, _ = UNet3DConditionModel.from_pretrained(
        OmegaConf.to_container(config.model),
        inference_ckpt_path,  # load checkpoint
        device="cpu",
    )
    unet = unet.to(dtype=dtype)
 
    # Optionally enable memory-efficient attention if available
    if is_xformers_available():
        unet.enable_xformers_memory_efficient_attention()
 
    # Initialize the pipeline and move to GPU
    pipeline = LipsyncPipeline(
        vae=vae,
        audio_encoder=audio_encoder,
        unet=unet,
        scheduler=scheduler,
    ).to("cuda")
 
    # Set seed
    if seed != -1:
        set_seed(seed)
    else:
        torch.seed()
 
    with autocast():
        try:
            pipeline(
                video_path=video_path,
                audio_path=audio_path,
                video_out_path=video_out_path,
                video_mask_path=video_out_path.replace(".mp4", "_mask.mp4"),
                num_frames=config.data.num_frames,
                num_inference_steps=inference_steps,
                guidance_scale=guidance_scale,
                weight_dtype=dtype,
                width=config.data.resolution,
                height=config.data.resolution,
            )
        finally:
            clear_cuda_memory()

def create_temp_dir():
    return tempfile.TemporaryDirectory()

def generate_audio(voice_cloning, text_prompt):
    if voice_cloning == 'yes':
        print('Entering Custom Audio creation using elevenlabs')
        set_api_key('92e149985ea2732b4359c74346c3daee')
        voice = Voice(voice_id="wu4wgNArkao4Vy9SnHzL",name="alex costa cacau",settings=VoiceSettings(
                        stability=0.71, similarity_boost=0.9, style=0.0, use_speaker_boost=True),)

        audio = generate(text = text_prompt, voice = voice, model = "eleven_multilingual_v2",stream=True, latency=4)
        with tempfile.NamedTemporaryFile(suffix=".mp3", prefix="cloned_audio_",dir=TEMP_DIR.name, delete=False) as temp_file:
            for chunk in audio:
                temp_file.write(chunk)
            driven_audio_path = temp_file.name
            print('driven_audio_path',driven_audio_path)

        return driven_audio_path

    elif voice_cloning == 'no':
        voice = 'echo'
        print('Entering Default Audio creation using elevenlabs')
        set_api_key('92e149985ea2732b4359c74346c3daee')
        audio = generate(text = text_prompt, voice = "Daniel", model = "eleven_multilingual_v2",stream=True, latency=4)
        with tempfile.NamedTemporaryFile(suffix=".mp3", prefix="default_audio_",dir=TEMP_DIR.name, delete=False) as temp_file:
            for chunk in audio:
                temp_file.write(chunk)
            driven_audio_path = temp_file.name
            print('driven_audio_path',driven_audio_path)
        return driven_audio_path

def get_video_duration(video_path):
    """Extracts video duration dynamically using ffprobe."""
    cmd = [
        "ffprobe", "-v", "error", "-show_entries", "format=duration", 
        "-of", "json", video_path
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    duration = json.loads(result.stdout)["format"]["duration"]
    return float(duration)

def extend_video_simple(video_path, audio_path, output_path):
    """Extends video duration by appending a reversed version if audio is longer."""
    audio_duration = librosa.get_duration(path=audio_path)
    video_duration = get_video_duration(video_path)

    print(f"Video Duration: {video_duration:.2f} sec")
    print(f"Audio Duration: {audio_duration:.2f} sec")

    if audio_duration > video_duration:
        print("Extending video by adding reversed version.")

        # Create a reversed version of the full video
        reversed_clip = tempfile.NamedTemporaryFile(dir=TEMP_DIR.name, delete=False, suffix=".mp4").name
        
        subprocess.run(
            f"ffmpeg -y -i {video_path} -vf reverse -an {reversed_clip}", shell=True
        )

        # Merge original + reversed
        subprocess.run(
            f"ffmpeg -y -i {video_path} -i {reversed_clip} -filter_complex \"[0:v:0][1:v:0]concat=n=2:v=1[outv]\" -map \"[outv]\" -an {output_path}",
            shell=True
        )
    else:
        print("Audio is not longer than video. No extension needed.")
        subprocess.run(f"cp {video_path} {output_path}", shell=True)

def extend_video_loop(video_path, audio_path, output_path):
    """Extends video duration by repeating original and reversed video until it meets/exceeds audio duration."""
    audio_duration = librosa.get_duration(path=audio_path)
    video_duration = get_video_duration(video_path)

    print(f"Video Duration: {video_duration:.2f} sec")
    print(f"Audio Duration: {audio_duration:.2f} sec")

    if audio_duration > video_duration:
        print("Extending video by repeating original and reversed versions.")

        # Create reversed video
        reversed_clip = tempfile.NamedTemporaryFile(dir=TEMP_DIR.name, delete=False, suffix=".mp4").name
        subprocess.run(
            f"ffmpeg -y -i {video_path} -vf reverse -an {reversed_clip}", shell=True
        )

        # Generate a list of clips to reach/exceed audio duration
        video_clips = [video_path, reversed_clip]
        total_duration = video_duration * 2  # Original + reversed

        while total_duration < audio_duration:
            video_clips.append(video_path)
            video_clips.append(reversed_clip)
            total_duration += video_duration * 2

        print(f"Total Clips: {len(video_clips)}")

        # Use FFmpeg filter_complex concat for seamless merging
        concat_filter = "".join(f"[{i}:v:0]" for i in range(len(video_clips))) + f"concat=n={len(video_clips)}:v=1[outv]"
        input_files = " ".join(f"-i {clip}" for clip in video_clips)

        subprocess.run(
            f"ffmpeg -y {input_files} -filter_complex \"{concat_filter}\" -map \"[outv]\" -an {output_path}",
            shell=True
        )

        print(f"Extended video saved to {output_path}")

    else:
        print("Audio is not longer than video. No extension needed.")
        subprocess.run(f"cp {video_path} {output_path}", shell=True)

def translate_text(text, target_language):
    if not text or text.strip() == "":
        return ""
    LANGUAGE_CODES = {"english": "en", "hindi": "hi"}   
    try:
        # Convert language name to code
        target_language_code = LANGUAGE_CODES.get(target_language.lower())
        
        # Use Google Translate with proper coroutine handling
        async def perform_translation():
            translator = Translator()
            result = await translator.translate(text, dest=target_language_code)
            return result.text if hasattr(result, 'text') else text
        
        # Run the async function in the event loop
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        result = loop.run_until_complete(perform_translation())
        loop.close()
        
        return result
    except Exception as e:
        print(f"Error translating text: {e}")
        # Return original text if translation fails
        return text

def generate_heygen_video(text_prompt, avatar_id=None, voice_id=None, background_color="#f6f6fc"):
    """Generate video using HeyGen API"""
    print("Generating video using HeyGen API...")
    
    # Default avatar and voice IDs if not provided
    default_avatar_id = "b2fe0a1b3393465db58ec15d92f69ef5"
    default_voice_id = "0503bcc31c3f43d895be188940fae86b"
    
    payload = {
        "caption": True,
        "title": "AI Avatar Video",
        "callback_id": f"avatar_{int(time.time())}",
        "dimension": {
            "width": 1280,
            "height": 720
        },
        "video_inputs": [
            {
                "character": {
                    "type": "avatar",
                    "avatar_id": avatar_id or default_avatar_id,
                    "talking_photo_id": "",
                    "scale": 1,
                    "avatar_style": "normal",
                    "offset": {
                        "x": 0,
                        "y": 0
                    },
                    "matting": True,
                    "circle_background_color": "#ffffff",
                    "talking_photo_style": "",
                    "talking_style": "stable",
                    "expression": "default",
                    "super_resolution": False
                },
                "voice": {
                    "type": "text",
                    "voice_id": voice_id or default_voice_id,
                    "input_text": text_prompt,
                    "speed": 1,
                    "pitch": 0,
                    "emotion": "Excited",
                    "locale": "en-US"
                },
                "background": {
                    "type": "color",
                    "value": background_color
                }
            }
        ]
    }

    headers = {
        "accept": "application/json",
        "content-type": "application/json",
        "x-api-key": HEYGEN_API_KEY
    }

    # Step 1: Send the video generation request
    response = requests.post(HEYGEN_GENERATE_URL, json=payload, headers=headers)
    
    try:
        response_data = response.json()
    except ValueError:
        raise Exception("Invalid response from HeyGen API")

    # Extract video_id from the response
    video_id = None
    if "data" in response_data:
        video_id = response_data["data"].get("video_id")
    else:
        video_id = response_data.get("video_id")

    if not video_id:
        raise Exception(f"No video_id returned from HeyGen API. Response: {response_data}")

    print(f"HeyGen video generation started with video_id: {video_id}")

    # Step 2: Poll for video status
    return poll_heygen_video_status(video_id, headers)

def poll_heygen_video_status(video_id, headers, poll_interval=10, max_attempts=60):
    """Poll HeyGen API for video completion status"""
    attempts = 0
    
    while attempts < max_attempts:
        try:
            params = {"video_id": video_id}
            status_response = requests.get(HEYGEN_STATUS_URL, headers=headers, params=params)
            
            if status_response.status_code == 200:
                status_data = status_response.json()
                status = status_data.get("data", {}).get("status")
                print(f"HeyGen polling attempt {attempts+1}: Video status is '{status}'")
                
                if status == "completed":
                    video_url = status_data.get("data", {}).get("video_url")
                    if video_url:
                        return download_heygen_video(video_url)
                    else:
                        raise Exception("Video completed but no video URL found")
                elif status == "failed":
                    error_msg = status_data.get("data", {}).get("error", "Unknown error")
                    raise Exception(f"HeyGen video generation failed: {error_msg}")
            else:
                print(f"Error polling HeyGen status: {status_response.status_code} - {status_response.text}")
                break
                
        except Exception as e:
            print(f"Exception during HeyGen polling: {e}")
            break
        
        attempts += 1
        time.sleep(poll_interval)
    
    raise Exception("HeyGen video generation timed out")

def download_heygen_video(video_url):
    """Download the completed video from HeyGen"""
    print(f"Downloading video from: {video_url}")
    
    try:
        response = requests.get(video_url, stream=True)
        response.raise_for_status()
        
        # Create a temporary file to store the downloaded video
        temp_video = tempfile.NamedTemporaryFile(dir=TEMP_DIR.name, delete=False, suffix="_heygen.mp4")
        
        with open(temp_video.name, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        
        print(f"HeyGen video downloaded to: {temp_video.name}")
        return temp_video.name
        
    except Exception as e:
        raise Exception(f"Failed to download HeyGen video: {e}")

@app.route('/run', methods=['POST'])
def generate_video():
    global TEMP_DIR
    TEMP_DIR = create_temp_dir()
    start_time = time.time()

    # Get form parameters
    text_prompt = request.form.get('text_prompt', '').strip()
    if not text_prompt:
        return jsonify({'error': 'Input text prompt cannot be blank'}), 400
    
    print('Input text prompt:', text_prompt)
    
    # Get processing parameters
    use_heygen = request.form.get('use_heygen', 'no').lower() == 'yes'
    voice_cloning = request.form.get('voice_cloning', 'no')
    target_language = request.form.get('target_language', 'original_text')
    chat_model_used = request.form.get('chat_model_used', 'ryzedb')
    app_id = request.form.get('app_id', '')
    
    # Validate app_id if using RyzeDB
    if chat_model_used == 'ryzedb' and not app_id:
        return jsonify({'error': 'App ID cannot be blank when using RyzeDB'}), 400

    try:
        # Process text prompt based on chat model selection
        if chat_model_used == 'ryzedb':
            start_time_ryze = time.time()
            print("Processing text with RyzeDB...")
            
            # Get response from RyzeDB
            ryze_response = ryzedb_chat_avatar(text_prompt, app_id)
            print("Response from RyzeDB inference:", ryze_response)
            
            # Clean up response if needed
            if "No information available" in ryze_response:
                ryze_response = re.sub(r'\\+', '', ryze_response)
            
            # Summarize with OpenAI
            openai_response = openai_chat_avatar(ryze_response)
            text_prompt = openai_response.choices[0].message.content.strip()
            
            end_time_ryze = time.time()
            ryze_processing_time = end_time_ryze - start_time_ryze
            print(f'Final processed text prompt using RyzeDB + OpenAI: {text_prompt}')
            print(f'Time to process with RyzeDB + OpenAI: {ryze_processing_time:.2f} seconds')
            
        elif chat_model_used == 'self':
            print("Using original text prompt without processing...")
            text_prompt = text_prompt.strip()
        else:
            print("Unknown chat model specified, using original text...")
            text_prompt = text_prompt.strip()
    
        # Translate text if needed
        if target_language != 'original_text':
            translated_text = translate_text(text_prompt, target_language)
            text_prompt = translated_text.strip()
            print('Translated input text prompt:', text_prompt)

        if use_heygen:
            print("Using HeyGen API for video generation...")
            
            # Get HeyGen-specific parameters
            avatar_id = request.form.get('heygen_avatar_id')
            voice_id = request.form.get('heygen_voice_id')
            background_color = request.form.get('background_color', '#f6f6fc')
            
            # Generate video using HeyGen
            final_output_video = generate_heygen_video(
                text_prompt=text_prompt,
                avatar_id=avatar_id,
                voice_id=voice_id,
                background_color=background_color
            )
            
        else:
            print("Using local AI avatar for video generation...")
            
            # Check if video file is provided for local processing
            if 'video' not in request.files:
                return jsonify({'error': 'Video file is required for local AI avatar processing.'}), 400
            
            video_file = request.files['video']
            
            # Generate audio using ElevenLabs
            temp_audio_path = generate_audio(voice_cloning, text_prompt)
            
            # Save uploaded video to temporary file
            with tempfile.NamedTemporaryFile(suffix=".mp4", prefix="input_", dir=TEMP_DIR.name, delete=False) as temp_file:
                temp_video_path = temp_file.name
                video_file.save(temp_video_path)
                print('temp_video_path:', temp_video_path)

            # Get inference parameters
            inference_ckpt_path = request.form.get('inference_ckpt_path', 'checkpoints/latentsync_unet.pt')
            unet_config_path = request.form.get('unet_config_path', 'configs/unet/second_stage.yaml')

            # Extend video to match audio duration
            output_video = tempfile.NamedTemporaryFile(dir=TEMP_DIR.name, delete=False, suffix=".mp4").name
            extend_video_loop(temp_video_path, temp_audio_path, output_video)
            
            # Generate final lip-sync video
            final_output_video = tempfile.NamedTemporaryFile(dir=TEMP_DIR.name, delete=False, suffix="_final_extended.mp4").name
            
            run_inference(
                video_path=output_video,
                audio_path=temp_audio_path,
                video_out_path=final_output_video,
                inference_ckpt_path=inference_ckpt_path,
                unet_config_path=unet_config_path,
                inference_steps=int(request.form.get('inference_steps', 20)),
                guidance_scale=float(request.form.get('guidance_scale', 1.0)),
                seed=int(request.form.get('seed', 1247))
            )

        # Save the final video to the videos directory
        if final_output_video and final_output_video.endswith('.mp4'):
            filename = f"avatar_video_{int(time.time())}.mp4"
            destination_path = os.path.join(VIDEO_DIRECTORY, filename)
            shutil.copy(final_output_video, destination_path)
            video_url = f"/videos/{filename}"
            
            processing_method = "HeyGen API" if use_heygen else "Local AI Avatar"
            
            # Calculate total processing time
            end_time = time.time()
            total_time = end_time - start_time
            
            return jsonify({
                "message": f"Video processed successfully using {processing_method}.",
                "output_video": video_url,
                "processing_method": processing_method,
                "text_prompt": text_prompt,
                "chat_model_used": chat_model_used,
                "time_taken": round(total_time, 2),
                "status": "success"
            }), 200
        else:
            return jsonify({'error': 'Failed to generate video'}), 500
            
    except Exception as e:
        print(f"Error generating video: {e}")
        return jsonify({'error': str(e)}), 500

@app.route("/videos/<string:filename>", methods=['GET'])
def serve_video(filename):
    return send_from_directory(VIDEO_DIRECTORY, filename, as_attachment=False)
 
@app.route("/health", methods=["GET"])
def health_status():
    response = {"online": "true"}
    return jsonify(response)

if __name__ == '__main__':
    app.run(debug=True)