app.py

import os
import re
import requests
import gradio as gr
from moviepy.editor import *
import edge_tts
import tempfile
import logging
from datetime import datetime
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
import nltk
from nltk.tokenize import sent_tokenize
from transformers import pipeline
import torch
import asyncio

# Configuración inicial
nltk.download('punkt', quiet=True)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Configuración de modelos
PEXELS_API_KEY = os.getenv("PEXELS_API_KEY")
MODEL_NAME = "DeepESP/gpt2-spanish"

# Solución robusta para obtener voces
async def get_voices():
    try:
        voices = await edge_tts.list_voices()
        voice_names = []
        for v in voices:
            try:
                name = v.get('Name', v.get('ShortName', 'Desconocido'))
                gender = v.get('Gender', 'Desconocido')
                locale = v.get('Locale', v.get('Language', 'Desconocido'))
                voice_names.append(f"{name} ({gender}, {locale})")
            except Exception as e:
                logger.warning(f"Error procesando voz: {v} - {str(e)}")
                continue
        return voice_names, voices
    except Exception as e:
        logger.error(f"Error al obtener voces: {str(e)}")
        return [], []

# Obtener voces de forma síncrona para la inicialización
import nest_asyncio
nest_asyncio.apply()

VOICE_NAMES, VOICES = [], []

async def get_voices():
    voces = await edge_tts.list_voices()
    voice_names = [f"{v['Name']} ({v['Gender']}, {v['LocaleName']})" for v in voces]
    return voice_names, voces

async def get_and_set_voices():
    global VOICE_NAMES, VOICES
    try:
        VOICE_NAMES, VOICES = await get_voices()
        if not VOICES:
            raise Exception("No se obtuvieron voces.")
    except Exception as e:
        logging.warning(f"No se pudieron cargar voces dinámicamente: {e}")
        VOICE_NAMES = ["Voz Predeterminada (Femenino, es-ES)"]
        VOICES = [{'ShortName': 'es-ES-ElviraNeural'}]

asyncio.get_event_loop().run_until_complete(get_and_set_voices())

def generar_guion_profesional(prompt):
    """Genera guiones con respaldo robusto"""
    try:
        generator = pipeline(
            "text-generation",
            model=MODEL_NAME,
            device=0 if torch.cuda.is_available() else -1
        )
        
        response = generator(
            f"Escribe un guion profesional para un video de YouTube sobre '{prompt}':\n\n1. Introducción\n2. Desarrollo\n3. Conclusión\n\n",
            max_length=800,
            temperature=0.7,
            num_return_sequences=1
        )
        
        return response[0]['generated_text']
        
    except Exception as e:
        logger.error(f"Error generando guion: {str(e)}")
        return f"""Guión de respaldo sobre {prompt}:
        
1. INTRODUCCIÓN: Hoy exploraremos {prompt}
2. DESARROLLO: Aspectos clave sobre el tema
3. CONCLUSIÓN: Resumen y cierre"""

def buscar_videos_avanzado(prompt, guion, num_videos=3):
    """Búsqueda con múltiples respaldos"""
    try:
        palabras = re.findall(r'\b\w{4,}\b', prompt.lower())[:5]
        response = requests.get(
            f"https://api.pexels.com/videos/search?query={'+'.join(palabras)}&per_page={num_videos}",
            headers={"Authorization": PEXELS_API_KEY},
            timeout=10
        )
        return response.json().get('videos', [])[:num_videos]
    except Exception as e:
        logger.error(f"Error buscando videos: {str(e)}")
        return []

async def crear_video_profesional(prompt, custom_script, voz_index, musica=None):
    try:
        # 1. Generar guión
        guion = custom_script if custom_script else generar_guion_profesional(prompt)
        
        # 2. Configurar voz
        voz_seleccionada = VOICES[voz_indimport os
import asyncio
from concurrent.futures import ThreadPoolExecutor
import gradio as gr

# Configuración CRÍTICA para evitar timeouts
GRADIO_TIMEOUT = 600  # 10 minutos (en segundos)
MAX_VIDEO_DURATION = 120  # 2 minutos (evita procesos eternos)

async def crear_video_profesional(prompt, custom_script, voz_index, musica=None):
    try:
        # 1. Simulamos un proceso largo (¡esto es lo que causa el timeout!)
        # Reemplaza esto con tu lógica real de generación
        await asyncio.sleep(30)  # Solo para prueba
        
        # 2. Devuelve un video de prueba (eliminar en producción)
        return "video_prueba.mp4"
        
    except Exception as e:
        print(f"ERROR: {str(e)}")
        return None

# 👇 **Solución Mágica**: Ejecución en hilos separados
def run_async_with_timeout(prompt, script, voz_index, musica=None):
    with ThreadPoolExecutor() as executor:
        future = executor.submit(
            lambda: asyncio.run(crear_video_profesional(prompt, script, voz_index, musica))
        )
        return future.result(timeout=GRADIO_TIMEOUT)

# Interfaz Minimalista (para enfocarnos en el timeout)
with gr.Blocks() as app:
    with gr.Row():
        prompt = gr.Textbox(label="Tema")
        btn = gr.Button("Generar")
    output = gr.Video()
    
    btn.click(
        fn=run_async_with_timeout,  # 👈 Usamos el wrapper anti-timeout
        inputs=[prompt, gr.Textbox(visible=False), gr.Number(visible=False)],
        outputs=output
    )

if __name__ == "__main__":
    app.launch(
        server_name="0.0.0.0",
        server_port=7860,
        # ⚠️ Configuración CLAVE para el timeout
        app_kwargs={"timeout": GRADIO_TIMEOUT}
    )