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)