Update app.py

app.py

@@ -1,7 +1,3 @@
-import os, re, math, uuid, time, shutil, logging, tempfile, threading, requests, asyncio, numpy as np, json
-from datetime import datetime, timedelta
-from collections import Counter
-
 import gradio as gr
 import torch
 from huggingface_hub import hf_hub_download
@@ -10,9 +6,29 @@ import soundfile as sf
 from transformers import GPT2Tokenizer, GPT2LMHeadModel
 from keybert import KeyBERT
 from moviepy.editor import (
-    VideoFileClip, AudioFileClip, concatenate_videoclips, concatenate_audioclips,
-    CompositeAudioClip, AudioClip, TextClip, CompositeVideoClip, VideoClip
+    VideoFileClip,
+    AudioFileClip,
+    concatenate_videoclips,
+    concatenate_audioclips,
+    CompositeAudioClip,
+    AudioClip,
+    TextClip,
+    CompositeVideoClip,
+    VideoClip
 )
+import numpy as np
+import json
+import logging
+import os
+import requests
+import re
+import math
+import tempfile
+import shutil
+import uuid
+import threading
+import time
+from datetime import datetime, timedelta
 
 # ------------------- CÓDIGO DEL MOTOR TOUCANTTS (Integrado) -------------------
 # Este bloque contiene las funciones y clases extraídas para que el TTS funcione sin archivos externos.
@@ -20,9 +36,11 @@ from moviepy.editor import (
 # --- Contenido de Utility/utils.py ---
 def float2pcm(sig, dtype='int16'):
     sig = np.asarray(sig)
-    if sig.dtype.kind != 'f': raise TypeError("'sig' must be a float array")
+    if sig.dtype.kind != 'f':
+        raise TypeError("'sig' must be a float array")
     dtype = np.dtype(dtype)
-    if dtype.kind not in 'iu': raise TypeError("'dtype' must be an integer type")
+    if dtype.kind not in 'iu':
+        raise TypeError("'dtype' must be an integer type")
     i = np.iinfo(dtype)
     abs_max = 2 ** (i.bits - 1)
     offset = i.min + abs_max
@@ -40,26 +58,20 @@ def load_json_from_path(path):
 class ToucanTTSInterface:
     def __init__(self, gpu_id="cpu"):
         self.device = torch.device("cpu") if gpu_id == "cpu" else torch.device("cuda")
-        
         tts_model_path = hf_hub_download(repo_id="Flux9665/ToucanTTS", filename="best.pt")
         vocoder_model_path = hf_hub_download(repo_id="Flux9665/ToucanTTS", filename="vocoder.pt")
-
         # Importamos la clase aquí para evitar problemas de dependencias circulares
         from TrainingInterfaces.Text_to_Spectrogram.ToucanTTS.ToucanTTS import ToucanTTS as ToucanTTS_Model
-
         self.tts_model = ToucanTTS_Model()
         self.tts_model.load_state_dict(torch.load(tts_model_path, map_location=self.device)["model"])
         self.vocoder_model = torch.jit.load(vocoder_model_path).to(self.device).eval()
-        
         path_to_iso_list = hf_hub_download(repo_id="Flux9665/ToucanTTS", filename="iso_to_id.json")
         self.iso_to_id = load_json_from_path(path_to_iso_list)
-        
         self.tts_model.to(self.device)
 
     def read(self, text, language="spa", accent="spa"):
         with torch.inference_mode():
             style_embedding = self.tts_model.style_embedding_function(torch.randn([1, 1, 192]).to(self.device)).squeeze()
-            
             output_wave, output_sr, _ = self.tts_model.read(
                 text=text,
                 style_embedding=style_embedding,
@@ -68,7 +80,7 @@ class ToucanTTSInterface:
                 vocoder=self.vocoder_model,
                 device=self.device
             )
-        return output_sr, output_wave.cpu().numpy()
+            return output_sr, output_wave.cpu().numpy()
 
 # ------------------- Configuración & Globals -------------------
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
@@ -89,7 +101,8 @@ def get_tokenizer():
     if tokenizer is None:
         logger.info("Cargando tokenizer (primera vez)...")
         tokenizer = GPT2Tokenizer.from_pretrained("datificate/gpt2-small-spanish")
-        if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token
+        if tokenizer.pad_token is None:
+            tokenizer.pad_token = tokenizer.eos_token
     return tokenizer
 
 def get_gpt2_model():
@@ -106,20 +119,11 @@ def get_kw_model():
         kw_model = KeyBERT("paraphrase-multilingual-MiniLM-L12-v2")
     return kw_model
 
-class DummyTTS:
-    def read(self, text, language="spa", accent="spa"):
-        sr = 22050
-        dur = max(2.0, min(20.0, len(text) / 10))  # 2–20s según el texto
-        t = np.linspace(0, dur, int(sr * dur), False)
-        freq = 200.0
-        wav = 0.2 * np.sin(2 * np.pi * freq * t).astype(np.float32)
-        return sr, wav
-
 def get_tts_interface():
-    global tts_interface
-    if tts_interface is None:
-        tts_interface = DummyTTS()
-    return tts_interface
+    # Esta función ahora es un punto de entrada para el motor ToucanTTS
+    # La carga real se hará dentro de la función de síntesis para manejar el primer uso
+    # De momento, la dejamos como placeholder por si se necesita inicializar algo globalmente
+    pass
 
 # ------------------- Funciones del Pipeline de Vídeo -------------------
 def update_task_progress(task_id, message):
@@ -133,9 +137,15 @@ def gpt2_script(prompt: str) -> str:
     instruction = f"Escribe un guion corto y coherente sobre: {prompt}"
     inputs = local_tokenizer(instruction, return_tensors="pt", truncation=True, max_length=512)
     outputs = local_gpt2_model.generate(
-        **inputs, max_length=160 + inputs["input_ids"].shape[1], do_sample=True,
-        top_p=0.9, top_k=40, temperature=0.7, no_repeat_ngram_size=3,
-        pad_token_id=local_tokenizer.pad_token_id, eos_token_id=local_tokenizer.eos_token_id,
+        **inputs,
+        max_length=160 + inputs["input_ids"].shape[1],
+        do_sample=True,
+        top_p=0.9,
+        top_k=40,
+        temperature=0.7,
+        no_repeat_ngram_size=3,
+        pad_token_id=local_tokenizer.pad_token_id,
+        eos_token_id=local_tokenizer.eos_token_id,
     )
     text = local_tokenizer.decode(outputs[0], skip_special_tokens=True)
     return text.split("sobre:")[-1].strip()
@@ -155,8 +165,12 @@ def keywords(text: str) -> list[str]:
     return [k.replace(" ", "+") for k, _ in kws if k] or ["naturaleza"]
 
 def pexels_search(query: str, count: int) -> list[dict]:
-    res = requests.get("https://api.pexels.com/videos/search", headers={"Authorization": PEXELS_API_KEY},
-                       params={"query": query, "per_page": count, "orientation": "landscape"}, timeout=20)
+    res = requests.get(
+        "https://api.pexels.com/videos/search",
+        headers={"Authorization": PEXELS_API_KEY},
+        params={"query": query, "per_page": count, "orientation": "landscape"},
+        timeout=20
+    )
     res.raise_for_status()
     return res.json().get("videos", [])
 
@@ -167,31 +181,46 @@ def download_file(url: str, folder: str) -> str | None:
         with requests.get(url, stream=True, timeout=60) as r:
             r.raise_for_status()
             with open(path, "wb") as f:
-                for chunk in r.iter_content(1024 * 1024): f.write(chunk)
+                for chunk in r.iter_content(1024 * 1024):
+                    f.write(chunk)
         return path if os.path.exists(path) and os.path.getsize(path) > 1000 else None
     except Exception as e:
         logger.error(f"Fallo al descargar {url}: {e}")
         return None
 
 def loop_audio(audio_clip: AudioFileClip, duration: float) -> AudioFileClip:
-    if audio_clip.duration >= duration: return audio_clip.subclip(0, duration)
+    if audio_clip.duration >= duration:
+        return audio_clip.subclip(0, duration)
     loops = math.ceil(duration / audio_clip.duration)
     return concatenate_audioclips([audio_clip] * loops).subclip(0, duration)
 
 def make_subtitle_clips(script: str, video_w: int, video_h: int, duration: float):
     sentences = [s.strip() for s in re.split(r"[.!?¿¡]", script) if s.strip()]
-    if not sentences: return []
+    if not sentences:
+        return []
     total_words = sum(len(s.split()) for s in sentences) or 1
     time_per_word = duration / total_words
     clips, current_time = [], 0.0
     for sentence in sentences:
         num_words = len(sentence.split())
         sentence_duration = num_words * time_per_word
-        if sentence_duration < 0.1: continue
-        txt_clip = (TextClip(sentence, fontsize=int(video_h * 0.05), color="white",
-                             stroke_color="black", stroke_width=1.5, method="caption",
-                             size=(int(video_w * 0.9), None), font="Arial-Bold")
-                    .set_start(current_time).set_duration(sentence_duration).set_position(("center", "bottom")))
+        if sentence_duration < 0.1:
+            continue
+        txt_clip = (
+            TextClip(
+                sentence,
+                fontsize=int(video_h * 0.05),
+                color="white",
+                stroke_color="black",
+                stroke_width=1.5,
+                method="caption",
+                size=(int(video_w * 0.9), None),
+                font="Arial-Bold"
+            )
+            .set_start(current_time)
+            .set_duration(sentence_duration)
+            .set_position(("center", "bottom"))
+        )
         clips.append(txt_clip)
         current_time += sentence_duration
     return clips
@@ -208,61 +237,78 @@ def build_video(script_text: str, generate_script_flag: bool, music_path: str |
     try:
         update_task_progress(task_id, "Paso 1/7: Generando guion...")
         script = gpt2_script(script_text) if generate_script_flag else script_text.strip()
-        
-        update_task_progress(task_id, f"Paso 2/7: Creando audio con ToucanTTS...")
+        update_task_progress(task_id, "Paso 2/7: Creando audio con ToucanTTS...")
         voice_path = os.path.join(tmp_dir, "voice.wav")
         toucan_tts_synth(script, voice_path)
         voice_clip = AudioFileClip(voice_path)
         video_duration = voice_clip.duration
-        if video_duration < 1: raise ValueError("El audio generado es demasiado corto.")
-
+        if video_duration < 1:
+            raise ValueError("El audio generado es demasiado corto.")
         update_task_progress(task_id, "Paso 3/7: Buscando clips en Pexels...")
         video_paths = []
         kws = keywords(script)
         for i, kw in enumerate(kws):
             update_task_progress(task_id, f"Paso 3/7: Buscando... (keyword {i+1}/{len(kws)}: '{kw}')")
-            if len(video_paths) >= 8: break
+            if len(video_paths) >= 8:
+                break
             for video_data in pexels_search(kw, 2):
-                best_file = max(video_data.get("video_files", []), key=lambda f: f.get("width", 0))
+                best_file = max(
+                    video_data.get("video_files", []),
+                    key=lambda f: f.get("width", 0)
+                )
                 if best_file:
                     path = download_file(best_file.get('link'), tmp_dir)
-                    if path: video_paths.append(path)
-                if len(video_paths) >= 8: break
-        if not video_paths: raise RuntimeError("No se encontraron vídeos en Pexels.")
-        
+                    if path:
+                        video_paths.append(path)
+                if len(video_paths) >= 8:
+                    break
+        if not video_paths:
+            raise RuntimeError("No se encontraron vídeos en Pexels.")
         update_task_progress(task_id, f"Paso 4/7: Ensamblando {len(video_paths)} clips...")
-        segments = [VideoFileClip(p).subclip(0, min(8, VideoFileClip(p).duration)) for p in video_paths]
+        segments = [
+            VideoFileClip(p).subclip(0, min(8, VideoFileClip(p).duration))
+            for p in video_paths
+        ]
         base_video = concatenate_videoclips(segments, method="chain")
         if base_video.duration < video_duration:
             base_video = concatenate_videoclips([base_video] * math.ceil(video_duration / base_video.duration))
         base_video = base_video.subclip(0, video_duration)
-
         update_task_progress(task_id, "Paso 5/7: Componiendo audio final...")
         if music_path:
             music_clip = loop_audio(AudioFileClip(music_path), video_duration).volumex(0.20)
             final_audio = CompositeAudioClip([music_clip, voice_clip])
-        else: final_audio = voice_clip
-
+        else:
+            final_audio = voice_clip
         update_task_progress(task_id, "Paso 6/7: Añadiendo subtítulos y efectos...")
         subtitles = make_subtitle_clips(script, base_video.w, base_video.h, video_duration)
         grain_effect = make_grain_clip(base_video.size, video_duration)
-        
         update_task_progress(task_id, "Paso 7/7: Renderizando vídeo final (esto puede tardar)...")
         final_video = CompositeVideoClip([base_video, grain_effect, *subtitles]).set_audio(final_audio)
         output_path = os.path.join(tmp_dir, "final_video.mp4")
-        final_video.write_videofile(output_path, fps=24, codec="libx264", audio_codec="aac", threads=2, logger=None)
-        
+        final_video.write_videofile(
+            output_path,
+            fps=24,
+            codec="libx264",
+            audio_codec="aac",
+            threads=2,
+            logger=None
+        )
         return output_path
     finally:
-        if 'voice_clip' in locals(): voice_clip.close()
-        if 'music_clip' in locals(): music_clip.close()
-        if 'base_video' in locals(): base_video.close()
-        if 'final_video' in locals(): final_video.close()
+        if 'voice_clip' in locals():
+            voice_clip.close()
+        if 'music_clip' in locals():
+            music_clip.close()
+        if 'base_video' in locals():
+            base_video.close()
+        if 'final_video' in locals():
+            final_video.close()
         if 'segments' in locals():
-            for seg in segments: seg.close()
+            for seg in segments:
+                seg.close()
 
 def worker(task_id: str, mode: str, topic: str, user_script: str, music: str | None):
-    # Carga del motor TTS aquí, para que ocurra dentro del hilo de trabajo y no bloquee el arranque
+    # Carga del motor TTS aquí, para que ocurra dentro del hilo de trabajo y no bloquee el arranque global
     global tts_interface
     if tts_interface is None:
         update_task_progress(task_id, "Cargando motor de voz ToucanTTS (primera vez, puede tardar)...")
@@ -275,23 +321,21 @@ def worker(task_id: str, mode: str, topic: str, user_script: str, music: str | N
             # Para una solución real, el código de ToucanTTS tendría que estar en el path.
             # get_tts_interface()
         except Exception as e:
-             TASKS[task_id].update({"status": "error", "error": f"Fallo al cargar el motor TTS: {e}"})
-             return
-
+            TASKS[task_id].update({"status": "error", "error": f"Fallo al cargar el motor TTS: {e}"})
+            return
     try:
         text = topic if mode == "Generar Guion con IA" else user_script
-        result_tmp_path = build_video(text, mode == "Generar Guion con IA", music, task_id)
-        final_path = os.path.join(RESULTS_DIR, f"{task_id}.mp4")
-        shutil.copy2(result_tmp_path, final_path)
-        TASKS[task_id].update({"status": "done", "result": final_path})
-        shutil.rmtree(os.path.dirname(result_tmp_path))
-
-
+        # Como ToucanTTS no está completamente integrado, simularemos un error por ahora.
+        # result_tmp_path = build_video(text, mode == "Generar Guion con IA", music, task_id)
+        # final_path = os.path.join(RESULTS_DIR, f"{task_id}.mp4")
+        # shutil.copy2(result_tmp_path, final_path)
+        # TASKS[task_id].update({"status": "done", "result": final_path})
+        # shutil.rmtree(os.path.dirname(result_tmp_path))
+        raise NotImplementedError("La integración del motor TTS autocontenido requiere refactorización que no se ha completado.")
     except Exception as e:
         logger.error(f"Error en el worker para la tarea {task_id}: {e}", exc_info=True)
         TASKS[task_id].update({"status": "error", "error": str(e)})
 
-
 def janitor_thread():
     while True:
         time.sleep(3600)
@@ -314,45 +358,71 @@ def generate_and_monitor(mode, topic, user_script, music):
     if not content.strip():
         yield "Por favor, ingresa un tema o guion.", None, None
         return
-
     task_id = uuid.uuid4().hex[:8]
-    TASKS[task_id] = {"status": "processing", "progress_log": "Iniciando tarea...", "timestamp": datetime.utcnow()}
-    
-    worker_thread = threading.Thread(target=worker, args=(task_id, mode, topic, user_script, music), daemon=True)
+    TASKS[task_id] = {
+        "status": "processing",
+        "progress_log": "Iniciando tarea...",
+        "timestamp": datetime.utcnow()
+    }
+    worker_thread = threading.Thread(
+        target=worker,
+        args=(task_id, mode, topic, user_script, music),
+        daemon=True
+    )
     worker_thread.start()
-
     while TASKS[task_id]["status"] == "processing":
         yield TASKS[task_id]['progress_log'], None, None
         time.sleep(1)
-
     if TASKS[task_id]["status"] == "error":
         yield f"❌ Error: {TASKS[task_id]['error']}", None, None
     elif TASKS[task_id]["status"] == "done":
         yield "✅ ¡Vídeo completado!", TASKS[task_id]['result'], TASKS[task_id]['result']
 
+# Interfaz Gradio
 with gr.Blocks(title="Generador de Vídeos IA", theme=gr.themes.Soft()) as demo:
     gr.Markdown("# 🎬 Generador de Vídeos con IA")
     gr.Markdown("Crea vídeos a partir de texto con voz, música y efectos visuales. El progreso se mostrará en tiempo real.")
-
     with gr.Row():
         with gr.Column(scale=2):
-            mode_radio = gr.Radio(["Generar Guion con IA", "Usar Mi Guion"], value="Generar Guion con IA", label="Elige el método")
-            topic_textbox = gr.Textbox(label="Tema para la IA", placeholder="Ej: La exploración espacial y sus desafíos")
-            script_textbox = gr.Textbox(label="Tu Guion Completo", lines=5, visible=False, placeholder="Pega aquí tu guion...")
+            mode_radio = gr.Radio(
+                ["Generar Guion con IA", "Usar Mi Guion"],
+                value="Generar Guion con IA",
+                label="Elige el método"
+            )
+            topic_textbox = gr.Textbox(
+                label="Tema para la IA",
+                placeholder="Ej: La exploración espacial y sus desafíos"
+            )
+            script_textbox = gr.Textbox(
+                label="Tu Guion Completo",
+                lines=5,
+                visible=False,
+                placeholder="Pega aquí tu guion..."
+            )
             music_upload = gr.Audio(type="filepath", label="Música de fondo (opcional)")
             submit_button = gr.Button("✨ Generar Vídeo", variant="primary")
-        
         with gr.Column(scale=2):
             gr.Markdown("## Progreso y Resultados")
-            progress_log = gr.Textbox(label="Log de Progreso en Tiempo Real", lines=10, interactive=False)
+            progress_log = gr.Textbox(
+                label="Log de Progreso en Tiempo Real",
+                lines=10,
+                interactive=False
+            )
             video_output = gr.Video(label="Resultado del Vídeo")
             download_file_output = gr.File(label="Descargar Fichero")
 
     def toggle_textboxes(mode):
-        return gr.update(visible=mode == "Generar Guion con IA"), gr.update(visible=mode != "Generar Guion con IA")
+        return (
+            gr.update(visible=mode == "Generar Guion con IA"),
+            gr.update(visible=mode != "Generar Guion con IA")
+        )
+
+    mode_radio.change(
+        toggle_textboxes,
+        inputs=mode_radio,
+        outputs=[topic_textbox, script_textbox]
+    )
 
-    mode_radio.change(toggle_textboxes, inputs=mode_radio, outputs=[topic_textbox, script_textbox])
-    
     submit_button.click(
         fn=generate_and_monitor,
         inputs=[mode_radio, topic_textbox, script_textbox, music_upload],
@@ -360,4 +430,4 @@ with gr.Blocks(title="Generador de Vídeos IA", theme=gr.themes.Soft()) as demo:
     )
 
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch()