Update app.py

app.py

@@ -16,10 +16,11 @@ import math
 from pydub import AudioSegment
 from pexelsapi.pexels import Pexels
 
-# Configuración inicial
+# 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")
 
 # Inicialización de modelos
@@ -29,138 +30,347 @@ try:
     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 cargado exitosamente.")
 except Exception as e:
     logger.error(f"Error al cargar modelo GPT-2: {e}")
-    tokenizer = model = None
+    tokenizer = None
+    model = None
 
 try:
     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 KeyBERT: {e}")
     kw_model = None
 
-# Funciones principales
-def generate_script(prompt, max_length=250):
+def generate_script(prompt, max_length=250): 
     if not tokenizer or not model:
-        return "Error: Modelo no disponible"
+        logger.error("Modelo GPT-2 no disponible")
+        return "Lo siento, el generador de guiones no está disponible."
     
+    logger.info("Generando guion con GPT-2...")
     try:
         inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512)
-        outputs = model.generate(**inputs, max_length=max_length)
-        return tokenizer.decode(outputs[0], skip_special_tokens=True)
+        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: {text[:200]}...")
+        return text
     except Exception as e:
-        logger.error(f"Error generando guión: {e}")
-        return "Error al generar guión"
+        logger.error(f"Error generando guion: {e}")
+        return "No se pudo generar el guion. Intenta con otro prompt."
 
-async def text_to_speech(text, output_path, voice="es-ES-ElviraNeural"):
+async def text_to_speech(text, voice="es-ES-ElviraNeural", output_path="voz.mp3"):
+    logger.info(f"Generando audio TTS...")
     try:
         communicate = edge_tts.Communicate(text, voice)
         await communicate.save(output_path)
+        logger.info(f"Audio TTS guardado")
         return True
     except Exception as e:
-        logger.error(f"Error en TTS: {e}")
+        logger.error(f"Error TTS: {e}")
         return False
 
-def search_pexels_videos(query_list, num_videos=3):
+def download_video_file(url, temp_dir): 
+    if not url:
+        return None
+    
+    file_name = url.split('/')[-1].split('?')[0]
+    if not file_name.endswith('.mp4'): 
+        file_name = f"video_temp_{os.getpid()}_{datetime.now().strftime('%f')}.mp4"
+    
+    output_path = os.path.join(temp_dir, file_name)
+    logger.info(f"Descargando video: {url}")
+    try:
+        response = requests.get(url, stream=True, timeout=30) 
+        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")
+        return output_path
+    except Exception as e:
+        logger.error(f"Error descargando video: {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)
+
+def extract_visual_keywords_from_script(script_text, max_keywords_per_segment=2):
+    if not kw_model:
+        logger.warning("KeyBERT no disponible. Usando método simple.")
+        return [script_text.split('.')[0].strip().replace(" ", "+")] if script_text.strip() else []
+    
+    logger.info("Extrayendo palabras clave...")
+    segments = [s.strip() for s in script_text.split('\n') if s.strip()]
+    if not segments: 
+        segments = [script_text]
+
+    all_keywords = set() 
+    for segment in segments:
+        if not segment: continue
+        try:
+            keywords_with_scores = kw_model.extract_keywords(
+                segment,
+                keyphrase_ngram_range=(1, 2), 
+                stop_words='spanish', 
+                top_n=max_keywords_per_segment,
+                use_mmr=True, 
+                diversity=0.7 
+            )
+            for kw, score in keywords_with_scores:
+                all_keywords.add(kw.replace(" ", "+"))
+        except Exception as e:
+            logger.warning(f"Error extrayendo keywords: {e}")
+            all_keywords.add(segment.split(' ')[0].strip().replace(" ", "+"))
+
+    return list(all_keywords)
+
+def search_pexels_videos(query_list, num_videos_per_query=5, min_duration_sec=7):
     if not PEXELS_API_KEY:
-        raise ValueError("API key no configurada")
+        logger.error("ERROR: PEXELS_API_KEY no configurada.")
+        raise ValueError("Configura PEXELS_API_KEY en los Secrets")
+    
+    if not query_list:
+        logger.warning("No hay queries para buscar.")
+        return []
     
     pexel = Pexels(PEXELS_API_KEY)
-    video_urls = []
+    all_video_urls = []
     
     for query in query_list:
+        logger.info(f"Buscando videos para: '{query}'")
         try:
-            results = pexel.search_videos(query=query, per_page=num_videos)
-            for video in results.get('videos', []):
-                best = max(video['video_files'], key=lambda x: x['width']*x['height'])
-                video_urls.append(best['link'])
+            results = pexel.search_videos(
+                query=query,
+                orientation='landscape',
+                per_page=num_videos_per_query
+            )
+            
+            videos = results.get('videos', [])
+            if not videos:
+                logger.info(f"No se encontraron videos para: '{query}'")
+                continue
+
+            for video in videos:
+                video_files = video.get('video_files', [])
+                if video_files:
+                    best_quality = max(
+                        video_files,
+                        key=lambda x: x.get('width', 0) * x.get('height', 0)
+                    )
+                    all_video_urls.append(best_quality['link'])
         except Exception as e:
             logger.error(f"Error buscando videos: {e}")
     
-    return video_urls
+    return all_video_urls
+
+def crear_video(prompt_type, input_text, musica_url=None):
+    logger.info(f"Iniciando creación de video: {prompt_type}")
+    guion = ""
+    if prompt_type == "Generar Guion con IA":
+        guion = generate_script(input_text)
+        if not guion or "No se pudo" in guion:
+            raise ValueError(guion) 
+    else: 
+        guion = input_text
+        if not guion.strip():
+            raise ValueError("Introduce tu guion.")
+
+    temp_files = []
+    downloaded_clip_paths = []
+    final_clips = []
+
+    temp_video_dir = tempfile.mkdtemp()
+    temp_files.append(temp_video_dir)
 
-def download_video(url, temp_dir):
     try:
-        response = requests.get(url, stream=True, timeout=30)
-        filename = f"video_{datetime.now().strftime('%Y%m%d%H%M%S')}.mp4"
-        filepath = os.path.join(temp_dir, filename)
+        voz_archivo = os.path.join(tempfile.gettempdir(), f"voz_temp_{os.getpid()}.mp3")
+        temp_files.append(voz_archivo)
+        if not asyncio.run(text_to_speech(guion, output_path=voz_archivo)):
+            raise ValueError("Error generando voz")
         
-        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
+        audio_tts = AudioFileClip(voz_archivo)
 
-def create_video(audio_path, video_paths, output_path):
-    try:
-        clips = [VideoFileClip(path) for path in video_paths]
-        final_clip = concatenate_videoclips(clips)
-        audio = AudioFileClip(audio_path)
+        search_queries = extract_visual_keywords_from_script(guion)
+        if not search_queries:
+            raise ValueError("No se pudieron extraer palabras clave")
+
+        video_urls = search_pexels_videos(search_queries)
+        if not video_urls:
+            raise ValueError(f"Pexels no encontró videos para: {search_queries}")
+
+        for url in video_urls:
+            path = download_video_file(url, temp_video_dir)
+            if path:
+                downloaded_clip_paths.append(path)
+
+        if not downloaded_clip_paths:
+            raise ValueError("No se pudo descargar videos")
+
+        total_desired_duration = audio_tts.duration * 1.2
+        current_duration = 0
         
-        if final_clip.duration < audio.duration:
-            final_clip = final_clip.loop(duration=audio.duration)
+        for path in downloaded_clip_paths:
+            try:
+                clip = VideoFileClip(path)
+                clip_duration = min(clip.duration, 10)
+                if clip_duration > 1:
+                    final_clips.append(clip.subclip(0, clip_duration))
+                    current_duration += clip_duration
+                    if current_duration >= total_desired_duration:
+                        break
+            except Exception as e:
+                logger.warning(f"Error procesando clip: {e}")
+
+        if not final_clips:
+            raise ValueError("No hay clips válidos")
+
+        video_base = concatenate_videoclips(final_clips, method="compose")
+        if video_base.duration < audio_tts.duration:
+            num_repeats = int(audio_tts.duration / video_base.duration) + 1
+            video_base = concatenate_videoclips([video_base] * num_repeats)
+
+        final_video_duration = audio_tts.duration
+        mezcla_audio = audio_tts
+
+        if musica_url and musica_url.strip():
+            musica_path = download_video_file(musica_url, temp_video_dir)
+            if musica_path:
+                temp_files.append(musica_path)
+                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)
+                    ])
+                except Exception as e:
+                    logger.warning(f"Error música: {e}")
+
+        video_final = video_base.set_audio(mezcla_audio).subclip(0, final_video_duration)
+        output_dir = "output_videos"
+        os.makedirs(output_dir, exist_ok=True)
+        output_path = os.path.join(output_dir, f"video_{datetime.now().strftime('%Y%m%d_%H%M%S')}.mp4")
         
-        final_clip = final_clip.set_audio(audio)
-        final_clip.write_videofile(
+        video_final.write_videofile(
             output_path,
+            fps=24,
+            threads=4,
             codec="libx264",
             audio_codec="aac",
-            fps=24
+            preset="medium",
+            ffmpeg_params=["-movflags", "+faststart"]
         )
-        return True
+        
+        return output_path
+
     except Exception as e:
-        logger.error(f"Error creando video: {e}")
-        return False
+        logger.error(f"Error general: {e}")
+        raise e
+    finally:
+        for f in temp_files:
+            if os.path.exists(f):
+                if os.path.isdir(f):
+                    import shutil
+                    shutil.rmtree(f)
+                else:
+                    os.remove(f)
+
+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("Introduce un tema para el guion.")
+    else:
+        input_text = prompt_manual
+        if not input_text.strip():
+            raise gr.Error("Introduce tu guion.")
+
+    try:
+        video_path = crear_video(prompt_type, input_text, musica_url if musica_url.strip() else None)
+        if video_path:
+            return video_path, "¡Video generado exitosamente!"
+        else:
+            raise gr.Error("Error desconocido")
+    except ValueError as ve:
+        return None, f"Error: {ve}"
+    except Exception as e:
+        return None, f"Error grave: {e}"
 
-# Interfaz de Gradio
 with gr.Blocks() as app:
-    with gr.Row():
-        with gr.Column():
-            prompt_type = gr.Radio(["Generar Guion", "Usar Guion Existente"], label="Tipo de Entrada")
-            text_input = gr.Textbox(label="Texto", lines=5)
-            generate_btn = gr.Button("Generar Video")
+    gr.Markdown("### 🎬 Generador de Video con Pexels")
+    
+    with gr.Tab("Generar Video"):
+        with gr.Row():
+            prompt_type = gr.Radio(
+                ["Generar Guion con IA", "Usar Mi Guion"], 
+                label="Método",
+                value="Generar Guion con IA"
+            )
+        
+        with gr.Column(visible=True) as ia_guion_column:
+            prompt_ia = gr.Textbox(
+                label="Tema para IA", 
+                lines=2
+            )
+        
+        with gr.Column(visible=False) as manual_guion_column:
+            prompt_manual = gr.Textbox(
+                label="Tu Guion", 
+                lines=5
+            )
+        
+        musica_input = gr.Textbox(
+            label="URL Música (opcional)", 
+        )
+        
+        boton = gr.Button("Generar Video")
         
         with gr.Column():
-            video_output = gr.Video()
-            status_output = gr.Textbox(label="Estado")
+            salida_video = gr.Video(label="Video Resultado", interactive=False)
+            estado_mensaje = gr.Textbox(label="Estado", interactive=False)
 
-    @generate_btn.click(inputs=[prompt_type, text_input], outputs=[video_output, status_output])
-    async def generate_video(prompt_type, text):
-        temp_dir = tempfile.mkdtemp()
-        try:
-            if prompt_type == "Generar Guion":
-                script = generate_script(text)
-            else:
-                script = text
-            
-            tts_path = os.path.join(temp_dir, "audio.mp3")
-            if not await text_to_speech(script, tts_path):
-                return None, "Error generando voz"
-            
-            keywords = extract_keywords(script) if kw_model else [text.split()[0]]
-            video_urls = search_pexels_videos(keywords)
-            
-            video_paths = []
-            for url in video_urls:
-                path = download_video(url, temp_dir)
-                if path:
-                    video_paths.append(path)
-            
-            if not video_paths:
-                return None, "Error descargando videos"
-            
-            output_path = os.path.join(temp_dir, "final.mp4")
-            if create_video(tts_path, video_paths, output_path):
-                return output_path, "Video generado"
-            else:
-                return None, "Error creando video"
-        except Exception as e:
-            logger.error(f"Error: {str(e)}")
-            return None, f"Error: {str(e)}"
-        finally:
-            pass  # Hugging Face limpia automáticamente
+    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, "Procesando..."),
+        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]
+    )
 
 if __name__ == "__main__":
     app.launch(server_name="0.0.0.0", server_port=7860)