Spaces:
Running
Running
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 | |
# --- Configuración de Logging --- | |
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
logger = logging.getLogger(__name__) | |
# --- Inicialización de Tokenizer y Modelo GPT-2 --- | |
# Especificamos un modelo más pequeño para una generación más rápida y menos exigente en recursos | |
MODEL_NAME = "gpt2-small" # Puedes cambiar a "gpt2" si tienes más RAM/GPU. | |
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}") | |
# En un entorno de producción, podrías querer un fallback o salir aquí. | |
tokenizer = None | |
model = None | |
# --- Funciones de Utilidad --- | |
def generate_script(prompt, max_length=250): # Max_length ajustado ligeramente | |
""" | |
Genera un guion usando el modelo GPT-2. | |
""" | |
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) | |
# Mover a GPU si está disponible | |
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"): | |
""" | |
Convierte texto a voz usando Edge TTS. | |
""" | |
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 # Relanzar la excepción para manejo en la función principal | |
def download_video_sample(url): | |
""" | |
Descarga un archivo de video desde una URL a un archivo temporal. | |
""" | |
if not url: # Manejar URL vacía | |
return None | |
logger.info(f"Intentando descargar video de ejemplo desde: {url}") | |
tmp = None # Inicializar tmp para asegurar que exista en caso de error | |
try: | |
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") | |
response = requests.get(url, stream=True, timeout=15) # Aumentar timeout | |
response.raise_for_status() # Lanza un error para códigos de estado HTTP malos | |
# Guardar el contenido en chunks | |
for chunk in response.iter_content(chunk_size=8192): # Chunk más pequeño | |
if chunk: # Filtrar chunks vacíos | |
tmp.write(chunk) | |
tmp.close() | |
logger.info(f"Video descargado a: {tmp.name}") | |
return tmp.name | |
except requests.exceptions.RequestException as e: | |
logger.error(f"Error de red/HTTP al descargar el video {url}: {e}") | |
if tmp and os.path.exists(tmp.name): | |
os.remove(tmp.name) | |
return None | |
except Exception as e: | |
logger.error(f"Error inesperado al descargar video {url}: {e}") | |
if tmp and os.path.exists(tmp.name): | |
os.remove(tmp.name) | |
return None | |
def loop_audio_to_length(audio_clip, target_duration): | |
""" | |
Buclea un clip de audio hasta una duración objetivo. | |
""" | |
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) | |
# --- Función Principal de Creación de Video --- | |
def crear_video(prompt_type, input_text, musica_url=None): | |
""" | |
Crea un video combinando un guion (generado o provisto), voz TTS, clips de video y música de fondo. | |
""" | |
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) # Propagar el error para Gradio | |
else: # prompt_type == "Usar Mi Guion" | |
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 = [] # Para llevar un registro de archivos temporales a limpiar | |
clips = [] # Lista para almacenar los objetos VideoFileClip | |
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. Descargar videos de ejemplo (¡REEMPLAZA ESTAS URLs CON TUS FUENTES REALES!) | |
# Sugerencia: Busca videos de stock gratuitos o crea un sistema para buscar por palabras clave | |
video_urls = [ | |
"https://www.learningcontainer.com/wp-content/uploads/2020/05/sample-mp4-file.mp4", | |
"https://file-examples.com/storage/fe2c91b5c46522c0734a74a/2017/04/file_example_MP4_480_1_5MG.mp4", | |
"https://sample-videos.com/video123/mp4/720/big_buck_bunny_720p_1mb.mp4", | |
"https://test-videos.co.uk/vids/bigbuckbunny/mp4/720/big_buck_bunny_720p_1mb.mp4" # Otra URL de ejemplo | |
] | |
valid_clip_found = False | |
for url in video_urls: | |
video_path = download_video_sample(url) | |
if video_path: | |
temp_files.append(video_path) | |
try: | |
# Limitar duración del clip individual para que no sea excesivo | |
clip = VideoFileClip(video_path).subclip(0, min(15, VideoFileClip(video_path).duration)) | |
if clip.duration > 1: # Asegurarse de que el clip tenga una duración mínima | |
clips.append(clip) | |
valid_clip_found = True | |
else: | |
logger.warning(f"Clip de video muy corto ({clip.duration:.2f}s) de {url}, omitiendo.") | |
clip.close() # Cerrar clip si no se va a usar | |
except Exception as e: | |
logger.warning(f"No se pudo cargar o procesar el clip de video {video_path} de {url}: {e}") | |
else: | |
logger.warning(f"No se pudo descargar el video de la URL: {url}") | |
if not valid_clip_found or not clips: | |
logger.error("No se pudieron obtener clips de video válidos. Abortando creación de video.") | |
raise ValueError("No se encontraron clips de video válidos. Asegúrate de que las URLs sean correctas y accesibles.") | |
# 3. Concatenar videos | |
video_base = concatenate_videoclips(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 | |
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") # No subclip aquí aún | |
# El video final tendrá la duración del audio combinado | |
final_video_duration = audio_tts.duration | |
# 4. Música de fondo en loop si está definida | |
mezcla_audio = audio_tts # Por defecto, solo la voz | |
if musica_url and musica_url.strip(): | |
musica_path = download_video_sample(musica_url) | |
if musica_path: | |
temp_files.append(musica_path) | |
try: | |
musica_audio = AudioFileClip(musica_path) | |
# Loop música hasta la duración final del video | |
musica_loop = loop_audio_to_length(musica_audio, final_video_duration) | |
# Mezclar audio TTS y música (TTS al 100%, música al 30%) | |
# Asegurarse de que el audio TTS tenga la duración correcta antes de mezclar | |
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.") | |
# 5. Asignar audio al video y ajustar duración del video final | |
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") | |
# 6. 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 # Re-lanzar para que Gradio lo muestre | |
finally: | |
# 7. Limpiar archivos temporales | |
for f in temp_files: | |
if os.path.exists(f): | |
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}") | |
# Cerrar los clips de moviepy explícitamente para liberar recursos | |
for clip in clips: | |
if clip: | |
clip.close() | |
if 'audio_tts' in locals() and audio_tts: | |
audio_tts.close() | |
if 'musica_audio' in locals() and musica_audio: | |
musica_audio.close() | |
if 'video_final' in locals() and video_final: | |
video_final.close() | |
if 'video_base' in locals() and video_base: | |
video_base.close() | |
def run_app(prompt_type, prompt_ia, prompt_manual, musica_url): | |
""" | |
Función envoltorio para Gradio que maneja la ejecución y los errores. | |
""" | |
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: # Usar Mi Guion | |
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: | |
# Esto se manejará mejor por las excepciones lanzadas en crear_video | |
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: {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 🚀 | |
Crea videos con guiones generados por IA o propios, voz automática y música de fondo. | |
""") | |
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="") | |
# Lógica para mostrar/ocultar columnas según el tipo de prompt | |
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] | |
) | |
# Conectar el botón a la función run_app | |
boton.click( | |
fn=lambda: (None, gr.update(value="Iniciando generación... Por favor, espera, esto puede tardar un minuto o más.")), | |
outputs=[salida_video, estado_mensaje], | |
queue=False # Desactivar cola para feedback inmediato | |
).then( | |
fn=run_app, | |
inputs=[prompt_type, prompt_ia, prompt_manual, musica_input], | |
outputs=[salida_video, estado_mensaje] | |
) | |
# Limpiar el mensaje de estado cuando el usuario cambia las entradas | |
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) # 'share=True' para un enlace público temporal (solo para pruebas) |