app.py

import os
import asyncio
import logging
import tempfile
import requests
from datetime import datetime
import edge_tts
import gradio as gr
import torch
import re
from keybert import KeyBERT

# Configuración básica de logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Clave API de Pexels (configurar en Secrets de Hugging Face)
PEXELS_API_KEY = os.environ.get("PEXELS_API_KEY", "YOUR_DEFAULT_API_KEY")

# Inicialización del modelo KeyBERT
try:
    kw_model = KeyBERT('distilbert-base-nli-mean-tokens')
    logger.info("Modelo KeyBERT cargado exitosamente.")
except Exception as e:
    logger.error(f"Error al cargar KeyBERT: {e}")
    kw_model = None

# --- Funciones principales optimizadas para Spaces ---

async def text_to_speech(text, output_path, voice="es-ES-ElviraNeural"):
    """Genera audio TTS usando edge-tts"""
    try:
        communicate = edge_tts.Communicate(text, voice)
        await communicate.save(output_path)
        return True
    except Exception as e:
        logger.error(f"Error en TTS: {e}")
        return False

def download_video(url, temp_dir):
    """Descarga un video desde una URL a un directorio temporal"""
    try:
        response = requests.get(url, stream=True, timeout=30)
        response.raise_for_status()
        
        filename = f"video_{datetime.now().strftime('%H%M%S%f')}.mp4"
        filepath = os.path.join(temp_dir, filename)
        
        with open(filepath, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
                
        return filepath
    except Exception as e:
        logger.error(f"Error descargando video: {e}")
        return None

def extract_keywords(text, max_keywords=3):
    """Extrae palabras clave usando KeyBERT o método simple como fallback"""
    if kw_model:
        try:
            keywords = kw_model.extract_keywords(
                text,
                keyphrase_ngram_range=(1, 2),
                top_n=max_keywords,
                use_mmr=True,
                diversity=0.7
            )
            return [kw[0].replace(" ", "+") for kw in keywords]
        except Exception as e:
            logger.warning(f"Error KeyBERT: {e}")
    
    # Fallback: método simple
    words = re.findall(r'\b\w+\b', text.lower())
    stop_words = {"el", "la", "los", "las", "de", "en", "y", "a", "que", "es", "por"}
    return list(set([w for w in words if len(w) > 3 and w not in stop_words][:max_keywords]))

def search_pexels_videos(query_list, per_query=2):
    """Busca videos en Pexels usando su API oficial"""
    if not PEXELS_API_KEY:
        logger.error("API_KEY de Pexels no configurada")
        return []
    
    headers = {"Authorization": PEXELS_API_KEY}
    video_urls = []
    
    for query in query_list:
        try:
            params = {
                "query": query,
                "per_page": per_query,
                "orientation": "landscape",
                "size": "medium"
            }
            
            response = requests.get(
                "https://api.pexels.com/videos/search",
                headers=headers,
                params=params,
                timeout=15
            )
            
            if response.status_code == 200:
                videos = response.json().get("videos", [])
                for video in videos:
                    video_files = video.get("video_files", [])
                    if video_files:
                        # Seleccionar el video con la mejor resolución
                        best_quality = max(
                            video_files, 
                            key=lambda x: x.get("width", 0) * x.get("height", 0)
                        )
                        video_urls.append(best_quality["link"])
        except Exception as e:
            logger.error(f"Error buscando videos: {e}")
    
    return video_urls

def create_video(audio_path, video_paths, output_path):
    """Crea el video final usando FFmpeg"""
    try:
        # Crear archivo de lista para concatenación
        with open("input_list.txt", "w") as f:
            for path in video_paths:
                f.write(f"file '{path}'\n")
        
        # Comando FFmpeg para concatenar videos y añadir audio
        cmd = [
            "ffmpeg", "-y",
            "-f", "concat",
            "-safe", "0",
            "-i", "input_list.txt",
            "-i", audio_path,
            "-c", "copy",
            "-shortest",
            output_path
        ]
        
        subprocess.run(cmd, check=True)
        return True
    except Exception as e:
        logger.error(f"Error creando video: {e}")
        return False

async def generate_video(text, music_url=None):
    """Función principal para generar el video"""
    temp_dir = tempfile.mkdtemp()
    all_files = []
    
    try:
        # 1. Generar audio TTS
        tts_path = os.path.join(temp_dir, "audio.mp3")
        if not await text_to_speech(text, tts_path):
            return None, "Error generando voz"
        all_files.append(tts_path)
        
        # 2. Extraer palabras clave
        keywords = extract_keywords(text)
        logger.info(f"Palabras clave identificadas: {keywords}")
        
        # 3. Buscar y descargar videos
        video_urls = search_pexels_videos(keywords)
        if not video_urls:
            return None, "No se encontraron videos para las palabras clave"
        
        video_paths = []
        for url in video_urls:
            path = download_video(url, temp_dir)
            if path:
                video_paths.append(path)
                all_files.append(path)
        
        if not video_paths:
            return None, "Error descargando videos"
        
        # 4. Crear video final
        output_path = os.path.join(temp_dir, "final_video.mp4")
        if create_video(tts_path, video_paths, output_path):
            return output_path, "Video creado exitosamente"
        else:
            return None, "Error en la creación del video"
            
    except Exception as e:
        logger.exception("Error inesperado")
        return None, f"Error: {str(e)}"
    finally:
        # Limpieza opcional (Hugging Face limpia automáticamente)
        pass

# --- Interfaz de Gradio ---
with gr.Blocks(title="Generador Automático de Videos con IA", theme="soft") as demo:
    gr.Markdown("# 🎬 Generador Automático de Videos con IA")
    gr.Markdown("Transforma texto en videos usando contenido de Pexels y voz sintetizada")
    
    with gr.Row():
        with gr.Column():
            text_input = gr.Textbox(
                label="Texto para el video",
                placeholder="Describe el contenido que quieres en el video...",
                lines=5
            )
            generate_btn = gr.Button("Generar Video", variant="primary")
        
        with gr.Column():
            video_output = gr.Video(label="Video Generado")
            status_output = gr.Textbox(label="Estado")
    
    generate_btn.click(
        fn=generate_video,
        inputs=[text_input],
        outputs=[video_output, status_output]
    )
    
    gr.Markdown("### Cómo funciona:")
    gr.Markdown("""
    1. Ingresa un texto descriptivo
    2. Nuestra IA extrae palabras clave
    3. Buscamos videos relacionados en Pexels
    4. Generamos voz con Edge TTS
    5. Combinamos todo en un video final
    """)

# Para Hugging Face Spaces
if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)