import os import asyncio import logging import tempfile import requests from datetime import datetime from moviepy.editor import VideoFileClip, concatenate_videoclips, AudioFileClip, CompositeAudioClip import edge_tts import gradio as gr from transformers import GPT2Tokenizer, GPT2LMHeadModel import torch # --- Importar la librería de Pexels --- from pexels_api import API # --- Importar KeyBERT para extracción de palabras clave --- from keybert import KeyBERT # Asegúrate de tener 'keybert' y 'sentence-transformers' en requirements.txt # --- Configuración de Logging --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # --- Clave API de Pexels --- PEXELS_API_KEY = os.environ.get("PEXELS_API_KEY") pexels_api = API(PEXELS_API_KEY) # --- Inicialización de Tokenizer y Modelo GPT-2 --- MODEL_NAME = "gpt2-small" try: tokenizer = GPT2Tokenizer.from_pretrained(MODEL_NAME) model = GPT2LMHeadModel.from_pretrained(MODEL_NAME).eval() if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token logger.info(f"Modelo GPT-2 '{MODEL_NAME}' cargado exitosamente.") except Exception as e: logger.error(f"Error al cargar el modelo GPT-2 '{MODEL_NAME}': {e}") tokenizer = None model = None # --- Inicialización del modelo KeyBERT --- try: # Se descarga la primera vez. 'multi-qa-MiniLM-L6-cos-v1' es un buen modelo multilingüe pequeño. # Considera 'distiluse-base-multilingual-cased-v2' para mejor calidad multilingüe si hay suficiente RAM. kw_model = KeyBERT('multi-qa-MiniLM-L6-cos-v1') logger.info("Modelo KeyBERT cargado exitosamente.") except Exception as e: logger.error(f"Error al cargar el modelo KeyBERT: {e}. La búsqueda de videos será menos precisa.", exc_info=True) kw_model = None # --- Funciones de Utilidad --- def generate_script(prompt, max_length=250): if not tokenizer or not model: logger.error("Modelo GPT-2 no disponible para generar guion.") return "Lo siento, el generador de guiones no está disponible en este momento." logger.info("Generando guion con GPT-2...") try: inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") inputs = {k: v.to(device) for k, v in inputs.items()} model.to(device) with torch.no_grad(): outputs = model.generate( **inputs, max_length=max_length, do_sample=True, top_p=0.95, top_k=60, temperature=0.9, pad_token_id=tokenizer.pad_token_id, eos_token_id=tokenizer.eos_token_id ) text = tokenizer.decode(outputs[0], skip_special_tokens=True) logger.info(f"Guion generado (longitud: {len(text)} caracteres): {text[:200]}...") return text except Exception as e: logger.error(f"Error durante la generación del guion: {e}") return "No se pudo generar el guion. Intenta con otro prompt o un guion propio." async def text_to_speech(text, voice="es-ES-ElviraNeural", output_path="voz.mp3"): logger.info(f"Generando audio TTS para: '{text[:100]}...'") try: communicate = edge_tts.Communicate(text, voice) await communicate.save(output_path) logger.info(f"Audio TTS guardado en {output_path}") except Exception as e: logger.error(f"Error al generar audio TTS: {e}") raise def download_video_file(url, temp_dir): # Renombrada y mejorada para descargar a un dir temporal if not url: return None # Intentar obtener el nombre del archivo de la URL o generar uno file_name = url.split('/')[-1].split('?')[0] if not file_name.endswith('.mp4'): # Asegurar extensión correcta file_name = f"video_temp_{os.getpid()}_{datetime.now().strftime('%f')}.mp4" output_path = os.path.join(temp_dir, file_name) logger.info(f"Intentando descargar video de: {url} a {output_path}") try: response = requests.get(url, stream=True, timeout=30) # Aumentar timeout por videos grandes response.raise_for_status() with open(output_path, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) logger.info(f"Video descargado a: {output_path}") return output_path except requests.exceptions.RequestException as e: logger.error(f"Error de red/HTTP al descargar el video {url}: {e}") if os.path.exists(output_path): os.remove(output_path) return None except Exception as e: logger.error(f"Error inesperado al descargar video {url}: {e}") if os.path.exists(output_path): os.remove(output_path) return None def loop_audio_to_length(audio_clip, target_duration): if audio_clip.duration >= target_duration: return audio_clip.subclip(0, target_duration) loops = int(target_duration / audio_clip.duration) + 1 audios = [audio_clip] * loops concatenated = concatenate_videoclips(audios) return concatenated.subclip(0, target_duration) # --- NUEVA LÓGICA: Extracción de palabras clave visuales del guion --- def extract_visual_keywords_from_script(script_text, max_keywords_per_segment=2): """ Extrae palabras clave visuales de un script dividiéndolo en segmentos. Retorna una lista de términos de búsqueda únicos. """ if not kw_model: logger.warning("Modelo KeyBERT no disponible. La extracción de palabras clave será limitada.") # Fallback si KeyBERT no carga: usar las primeras palabras del guion return [script_text.split('.')[0].strip().replace(" ", "+")] if script_text.strip() else [] logger.info("Extrayendo palabras clave visuales del guion.") # Dividir el guion en frases o párrafos para un mejor análisis contextual # Puedes ajustar esto, por ejemplo, split('.') si prefieres por frases. segments = [s.strip() for s in script_text.split('\n') if s.strip()] if not segments: # Si no se puede dividir, usar el texto completo segments = [script_text] all_keywords = set() # Usar un set para evitar duplicados for segment in segments: if not segment: continue try: # keyphrase_ngram_range=(1,2) busca palabras sueltas o frases de dos palabras # top_n=max_keywords_per_segment es cuántas palabras clave por segmento # use_mmr y diversity ayudan a obtener palabras clave más variadas. keywords_with_scores = kw_model.extract_keywords( segment, keyphrase_ngram_range=(1, 2), stop_words='spanish', # ¡IMPORTANTE! Usar stop words en español top_n=max_keywords_per_segment, use_mmr=True, diversity=0.7 ) for kw, score in keywords_with_scores: # Convertir a formato de búsqueda de URL (espacios por +) all_keywords.add(kw.replace(" ", "+")) except Exception as e: logger.warning(f"Error al extraer palabras clave de un segmento: {e}") # En caso de error en un segmento, intentar con el segmento completo como palabra clave simple all_keywords.add(segment.split(' ')[0].strip().replace(" ", "+")) logger.info(f"Palabras clave visuales extraídas del guion: {list(all_keywords)}") return list(all_keywords) # Devolver como lista para iterar # --- Búsqueda de videos en Pexels --- def search_pexels_videos(query_list, num_videos_per_query=5, min_duration_sec=7): """ Busca videos en Pexels basados en una lista de queries y devuelve URLs de descarga. """ if not PEXELS_API_KEY: logger.error("ERROR: PEXELS_API_KEY no está configurada. No se pueden buscar videos en Pexels.") raise ValueError("PEXELS_API_KEY no está configurada. Por favor, configúrala como un secreto en Hugging Face Spaces.") if not query_list: logger.warning("Lista de queries para Pexels vacía. No se buscarán videos.") return [] all_video_urls = set() # Usar un set para evitar URLs duplicadas for query in query_list: logger.info(f"Buscando {num_videos_per_query} videos en Pexels para la query: '{query}'") try: # Pedimos más de los que necesitamos por query por si algunos no cumplen duración pexels_api.search_videos(query, per_page=num_videos_per_query * 2) videos = pexels_api.get_videos() if not videos: logger.info(f"No se encontraron videos en Pexels para la query: '{query}'") continue for video in videos: if video.get('duration') and video['duration'] < min_duration_sec: # logger.info(f"Saltando video {video['id']} de Pexels por duración ({video['duration']}s < {min_duration_sec}s).") continue best_quality_url = None for file in video.get('video_files', []): if file.get('quality') in ['hd', 'sd', 'medium', 'large'] and file.get('link'): best_quality_url = file['link'] break elif file.get('link'): best_quality_url = file['link'] if best_quality_url: all_video_urls.add(best_quality_url) else: logger.warning(f"No se encontró URL de descarga válida para el video Pexels ID: {video.get('id')}") except Exception as e: logger.error(f"Error al buscar videos en Pexels para '{query}': {e}", exc_info=True) # No lanzamos la excepción aquí, para que otras queries puedan seguir funcionando # Pero el error se registra. final_urls = list(all_video_urls) logger.info(f"Total de {len(final_urls)} URLs de video únicas obtenidas de Pexels.") return final_urls # --- Función Principal de Creación de Video --- def crear_video(prompt_type, input_text, musica_url=None): logger.info(f"Iniciando creación de video. Tipo de prompt: {prompt_type}") guion = "" if prompt_type == "Generar Guion con IA": guion = generate_script(input_text) if not guion or guion == "No se pudo generar el guion. Intenta con otro prompt o un guion propio.": raise ValueError(guion) else: guion = input_text if not guion.strip(): raise ValueError("Por favor, introduce tu guion.") if not guion.strip(): raise ValueError("El guion está vacío. No se puede proceder.") temp_files = [] downloaded_clip_paths = [] # Para los paths de los videos descargados final_clips = [] # Para los objetos VideoFileClip ya procesados # Directorio temporal para descargar videos temp_video_dir = tempfile.mkdtemp() temp_files.append(temp_video_dir) # Añadir el directorio temporal para su limpieza try: # 1. Generar audio TTS voz_archivo = os.path.join(tempfile.gettempdir(), f"voz_temp_{os.getpid()}.mp3") temp_files.append(voz_archivo) asyncio.run(text_to_speech(guion, output_path=voz_archivo)) audio_tts = AudioFileClip(voz_archivo) # 2. Extracción de palabras clave y búsqueda masiva en Pexels # Dividimos el guion en segmentos y extraemos palabras clave de cada uno # Esto genera una lista de queries para Pexels, como ['misterios+china', 'muralla+china', ...] search_queries_for_pexels = extract_visual_keywords_from_script(guion, max_keywords_per_segment=2) # Si no se pudo extraer ninguna palabra clave útil, lanzar error. if not search_queries_for_pexels: raise ValueError("No se pudieron extraer palabras clave visuales del guion para buscar videos.") # Buscar y obtener URLs de muchos videos de Pexels (e.g., 5 videos por cada palabra clave) pexels_video_urls_found = search_pexels_videos(search_queries_for_pexels, num_videos_per_query=5, min_duration_sec=7) if not pexels_video_urls_found: logger.error("Pexels no devolvió ningún video para las palabras clave extraídas.") raise ValueError(f"Pexels no encontró videos adecuados para el guion. Intenta con otro tema o guion más descriptivo. Palabras clave usadas: {search_queries_for_pexels}") # 3. Descargar todos los videos encontrados de Pexels logger.info(f"Descargando {len(pexels_video_urls_found)} videos de Pexels...") for url in pexels_video_urls_found: video_path = download_video_file(url, temp_video_dir) # Descargar al directorio temporal if video_path: downloaded_clip_paths.append(video_path) else: logger.warning(f"No se pudo descargar video de Pexels: {url}") if not downloaded_clip_paths: logger.error("No se pudo descargar ningún video válido de Pexels.") raise ValueError("No se pudo descargar ningún video válido de Pexels. Revisa la conexión o la calidad de las URLs.") # 4. Cargar y procesar los clips descargados total_desired_video_duration = audio_tts.duration * 1.2 # Intentar tener 20% más de video que audio current_video_clips_duration = 0 for path in downloaded_clip_paths: try: clip = VideoFileClip(path) # Opcional: Si quieres clips cortos y variados para transiciones rápidas clip_duration = min(clip.duration, 10) # Limitar a un máximo de 10 segundos por clip if clip_duration > 1: # Asegurar que el clip sea lo suficientemente largo final_clips.append(clip.subclip(0, clip_duration)) current_video_clips_duration += clip_duration if current_video_clips_duration >= total_desired_video_duration: break # Ya tenemos suficiente metraje else: logger.warning(f"Clip de video muy corto ({clip_duration:.2f}s) de {path}, omitiendo.") clip.close() except Exception as e: logger.warning(f"No se pudo cargar o procesar el clip de video {path}: {e}. Omitiendo.") if 'clip' in locals() and clip: clip.close() # Asegurarse de cerrar si hay un error if not final_clips: logger.error("No se pudo obtener ningún clip de video usable después de la descarga y procesamiento.") raise ValueError("No se pudieron procesar los videos descargados de Pexels.") # 5. Concatenar videos y ajustar duración video_base = concatenate_videoclips(final_clips, method="compose") logger.info(f"Video base concatenado, duración: {video_base.duration:.2f}s") # Asegurarse de que el video base sea al menos tan largo como el audio TTS # Si es más corto, loopearlo. if video_base.duration < audio_tts.duration: logger.info(f"Duración del video ({video_base.duration:.2f}s) es menor que la del audio TTS ({audio_tts.duration:.2f}s). Repitiendo video.") num_repeats = int(audio_tts.duration / video_base.duration) + 1 repeated_clips = [video_base] * num_repeats video_base = concatenate_videoclips(repeated_clips, method="compose") # El video final siempre tendrá la duración del audio TTS (o la mezcla de audio) final_video_duration = audio_tts.duration # 6. Música de fondo en loop mezcla_audio = audio_tts if musica_url and musica_url.strip(): musica_path = download_video_file(musica_url, temp_video_dir) # Usar la misma función de descarga if musica_path: temp_files.append(musica_path) # Añadir el path del archivo de música a limpiar try: musica_audio = AudioFileClip(musica_path) musica_loop = loop_audio_to_length(musica_audio, final_video_duration) mezcla_audio = CompositeAudioClip([musica_loop.volumex(0.3), audio_tts.set_duration(final_video_duration).volumex(1.0)]) logger.info("Música de fondo añadida y mezclada.") except Exception as e: logger.warning(f"No se pudo procesar la música de fondo de {musica_url}: {e}. Se usará solo la voz.") else: logger.warning(f"No se pudo descargar la música de {musica_url}. Se usará solo la voz.") # 7. Asignar audio al video y ajustar duración video_final = video_base.set_audio(mezcla_audio).subclip(0, final_video_duration) logger.info(f"Video final configurado con audio. Duración final: {video_final.duration:.2f}s") # 8. Guardar video final output_dir = "generated_videos" os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, f"video_output_{datetime.now().strftime('%Y%m%d_%H%M%S')}.mp4") logger.info(f"Escribiendo video final a: {output_path}") video_final.write_videofile(output_path, fps=24, threads=4, logger=None, preset="medium", codec="libx264") logger.info(f"Video generado exitosamente en: {output_path}") return output_path except Exception as e: logger.error(f"Error general en la creación del video: {e}", exc_info=True) raise e finally: # 9. Limpiar archivos y directorios temporales for f in temp_files: if os.path.exists(f): if os.path.isdir(f): try: import shutil shutil.rmtree(f) logger.info(f"Directorio temporal eliminado: {f}") except Exception as e: logger.warning(f"No se pudo eliminar el directorio temporal {f}: {e}") else: try: os.remove(f) logger.info(f"Archivo temporal eliminado: {f}") except Exception as e: logger.warning(f"No se pudo eliminar el archivo temporal {f}: {e}") # Asegurarse de cerrar todos los objetos MoviePy VideoFileClip y AudioFileClip for clip in final_clips: # Iterar sobre los clips finales que se usaron if clip and not clip.is_playing: # Solo cerrar si no está ya cerrado clip.close() if 'audio_tts' in locals() and audio_tts and not audio_tts.is_playing: audio_tts.close() if 'musica_audio' in locals() and musica_audio and not musica_audio.is_playing: musica_audio.close() if 'video_final' in locals() and video_final and not video_final.is_playing: video_final.close() if 'video_base' in locals() and video_base and not video_base.is_playing: video_base.close() def run_app(prompt_type, prompt_ia, prompt_manual, musica_url): input_text = "" if prompt_type == "Generar Guion con IA": input_text = prompt_ia if not input_text.strip(): raise gr.Error("Por favor, introduce un tema para generar el guion.") else: input_text = prompt_manual if not input_text.strip(): raise gr.Error("Por favor, introduce tu guion.") logger.info(f"Solicitud recibida: Tipo='{prompt_type}', Input='{input_text[:50]}...', Música='{musica_url}'") try: video_path = crear_video(prompt_type, input_text, musica_url if musica_url.strip() else None) if video_path: logger.info(f"Proceso completado. Video disponible en: {video_path}") return video_path, gr.update(value="¡Video generado exitosamente!") else: raise gr.Error("Hubo un problema desconocido al generar el video. Revisa los logs.") except ValueError as ve: logger.error(f"Error de validación o búsqueda de Pexels: {ve}") return None, gr.update(value=f"Error: {ve}", text_color="red") except Exception as e: logger.error(f"Error inesperado al ejecutar la aplicación: {e}", exc_info=True) return None, gr.update(value=f"Ocurrió un error grave: {e}. Por favor, inténtalo de nuevo.", text_color="red") # --- Interfaz de Gradio --- with gr.Blocks() as app: gr.Markdown(""" ### 🎬 Generador de Video Inteligente con Pexels 🚀 Crea videos con guiones generados por IA o propios, voz automática y **videos de stock relevantes de Pexels (sin videos de ejemplo)**. """) with gr.Tab("Generar Video"): with gr.Row(): prompt_type = gr.Radio( ["Generar Guion con IA", "Usar Mi Guion"], label="Método de Guion", value="Generar Guion con IA" ) with gr.Column(visible=True) as ia_guion_column: prompt_ia = gr.Textbox( label="Tema para Generar Guion (con IA)", lines=2, placeholder="Ej: Las maravillas del universo y las estrellas." ) with gr.Column(visible=False) as manual_guion_column: prompt_manual = gr.Textbox( label="Introduce Tu Guion Propio", lines=5, placeholder="Ej: Hola a todos, hoy hablaremos de un tema fascinante..." ) musica_input = gr.Textbox( label="URL de Música de Fondo (opcional, MP3 recomendado)", placeholder="Ej: https://www.soundhelix.com/examples/mp3/SoundHelix-Song-1.mp3" ) boton = gr.Button("✨ Generar Video") with gr.Column(): salida_video = gr.Video(label="Video Generado", interactive=False) estado_mensaje = gr.Textbox(label="Estado del Proceso", interactive=False, value="") prompt_type.change( fn=lambda value: (gr.update(visible=value == "Generar Guion con IA"), gr.update(visible=value == "Usar Mi Guion")), inputs=prompt_type, outputs=[ia_guion_column, manual_guion_column] ) boton.click( fn=lambda: (None, gr.update(value="Iniciando generación... Esto puede tardar varios minutos (descarga de videos).")), outputs=[salida_video, estado_mensaje], queue=False ).then( fn=run_app, inputs=[prompt_type, prompt_ia, prompt_manual, musica_input], outputs=[salida_video, estado_mensaje] ) prompt_ia.change(fn=lambda: gr.update(value=""), outputs=estado_mensaje, queue=False) prompt_manual.change(fn=lambda: gr.update(value=""), outputs=estado_mensaje, queue=False) musica_input.change(fn=lambda: gr.update(value=""), outputs=estado_mensaje, queue=False) if __name__ == "__main__": logger.info("Iniciando aplicación Gradio para Hugging Face Spaces...") app.launch(server_name="0.0.0.0", server_port=7860, share=False)