Update app.py

app.py

@@ -1,335 +1,237 @@
-import os
-import asyncio
-import logging
-import tempfile
-import requests
-from datetime import datetime
-import edge_tts
+import os, re, math, uuid, time, shutil, logging, tempfile, threading, requests, numpy as np
+from datetime import datetime, timedelta
+from collections import Counter
+
 import gradio as gr
 import torch
 from transformers import GPT2Tokenizer, GPT2LMHeadModel
 from keybert import KeyBERT
-from moviepy.editor import VideoFileClip, concatenate_videoclips, AudioFileClip, CompositeAudioClip
-import re
-import json
-import uuid
-import threading
-from queue import Queue
-import time
+from TTS.api import TTS
+from moviepy.editor import (
+    VideoFileClip, AudioFileClip, concatenate_videoclips, concatenate_audioclips,
+    CompositeAudioClip, AudioClip, TextClip, CompositeVideoClip, VideoClip, vfx
+)
 
-# Configuración de logging
-logging.basicConfig(level=logging.INFO)
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
 logger = logging.getLogger(__name__)
 
-# Directorio persistente para archivos
-PERSIST_DIR = "persistent_data"
-os.makedirs(PERSIST_DIR, exist_ok=True)
-
-# Cola de procesamiento
-processing_queue = Queue()
-task_status = {}
+PEXELS_API_KEY = os.getenv("PEXELS_API_KEY")
+if not PEXELS_API_KEY:
+    raise RuntimeError("Debes definir PEXELS_API_KEY en Variables & secrets")
+
+tokenizer = GPT2Tokenizer.from_pretrained("datificate/gpt2-small-spanish")
+gpt2 = GPT2LMHeadModel.from_pretrained("datificate/gpt2-small-spanish").eval()
+if tokenizer.pad_token is None:
+    tokenizer.pad_token = tokenizer.eos_token
+kw_model = KeyBERT("distilbert-base-multilingual-cased")
+tts_engine = TTS(model_name="tts_models/es/css10/vits", progress_bar=False, gpu=False)
+
+RESULTS_DIR = "video_results"
+os.makedirs(RESULTS_DIR, exist_ok=True)
+TASKS = {}
+
+# ───────── helpers ────────────────────────────────────────────────────────────────
+def gpt2_script(prompt: str, mx: int = 160) -> str:
+    ins = f"Escribe un guion corto, interesante y coherente sobre: {prompt}"
+    inp = tokenizer(ins, return_tensors="pt", truncation=True, max_length=512)
+    out = gpt2.generate(
+        **inp, max_length=mx + inp["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=tokenizer.pad_token_id, eos_token_id=tokenizer.eos_token_id,
+    )
+    txt = tokenizer.decode(out[0], skip_special_tokens=True)
+    return txt.split("sobre:")[-1].strip()[:mx]
 
-# Clase para manejar tareas
-class VideoTask:
-    def __init__(self, task_id, prompt_type, input_text, musica_file=None):
-        self.task_id = task_id
-        self.prompt_type = prompt_type
-        self.input_text = input_text
-        self.musica_file = musica_file
-        self.status = "pending"
-        self.progress = 0
-        self.result = None
-        self.error = None
-        self.steps_completed = []
-        
-    def to_dict(self):
-        return {
-            "task_id": self.task_id,
-            "status": self.status,
-            "progress": self.progress,
-            "result": self.result,
-            "error": self.error,
-            "steps_completed": self.steps_completed
-        }
+def coqui_tts(text: str, path: str):
+    text = re.sub(r"[^\w\s.,!?áéíóúñÁÉÍÓÚÑ]", "", text)[:500]
+    tts_engine.tts_to_file(text=text, file_path=path)
 
-# Worker thread para procesar videos
-def video_processor_worker():
-    while True:
-        try:
-            task = processing_queue.get(timeout=1)
-            if task is None:
+def keywords(text: str) -> list[str]:
+    clean = re.sub(r"[^\w\sáéíóúñÁÉÍÓÚÑ]", "", text.lower())
+    try:
+        kws = kw_model.extract_keywords(clean, stop_words="spanish", top_n=5)
+        return [k.replace(" ", "+") for k, _ in kws if k]
+    except Exception:
+        words = [w for w in clean.split() if len(w) > 4]
+        return [w for w, _ in Counter(words).most_common(5)] or ["nature"]
+
+def pexels_search(q: str, n: int) -> list[dict]:
+    r = requests.get(
+        "https://api.pexels.com/videos/search",
+        headers={"Authorization": PEXELS_API_KEY},
+        params={"query": q, "per_page": n, "orientation": "landscape"},
+        timeout=20,
+    )
+    r.raise_for_status()
+    return r.json().get("videos", [])
+
+def download(url: str, folder: str) -> str | None:
+    name = uuid.uuid4().hex + ".mp4"
+    path = os.path.join(folder, name)
+    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)
+    return path if os.path.getsize(path) > 1000 else None
+
+def loop_audio(aclip: AudioFileClip, dur: float) -> AudioFileClip:
+    if aclip.duration >= dur:
+        return aclip.subclip(0, dur)
+    loops = math.ceil(dur / aclip.duration)
+    return concatenate_audioclips([aclip] * loops).subclip(0, dur)
+
+def make_subs_clips(script: str, video_w: int, video_h: int, duration: float):
+    sentences = [s.strip() for s in re.split(r"[.!?¿¡]", script) if s.strip()]
+    total_words = sum(len(s.split()) for s in sentences) or 1
+    word_time = duration / total_words
+    clips, cursor = [], 0.0
+    for sent in sentences:
+        n_words = len(sent.split())
+        dur = n_words * word_time
+        txt_clip = (
+            TextClip(sent, fontsize=int(video_h * 0.05), color="white",
+                     stroke_color="black", stroke_width=2, method="caption",
+                     size=(int(video_w * 0.9), None))
+            .set_start(cursor)
+            .set_duration(dur)
+            .set_position(("center", video_h * 0.85))
+        )
+        clips.append(txt_clip)
+        cursor += dur
+    return clips
+
+def make_grain_clip(size: tuple[int, int], duration: float):
+    w, h = size
+    def frame(_t):
+        noise = np.random.randint(0, 256, (h, w, 1), dtype=np.uint8)
+        return np.repeat(noise, 3, axis=2)
+    return VideoClip(frame, duration=duration).set_opacity(0.15)
+
+# ───────── video builder ──────────────────────────────────────────────────────────
+def build_video(text: str, gen_script: bool, music_fp: str | None) -> str:
+    tmp = tempfile.mkdtemp()
+    script = gpt2_script(text) if gen_script else text.strip()
+    voice_path = os.path.join(tmp, "voice.mp3")
+    coqui_tts(script, voice_path)
+    voice_clip = AudioFileClip(voice_path)
+    adur = voice_clip.duration
+
+    vids = []
+    for kw in keywords(script):
+        if len(vids) >= 8:
+            break
+        for v in pexels_search(kw, 2):
+            best = max(v["video_files"], key=lambda x: x["width"] * x["height"])
+            p = download(best["link"], tmp)
+            if p:
+                vids.append(p)
+            if len(vids) >= 8:
                 break
-                
-            logger.info(f"Procesando tarea: {task.task_id}")
-            process_video_task(task)
-            
-        except:
-            time.sleep(0.1)
-            continue
-
-# Iniciar worker thread
-worker_thread = threading.Thread(target=video_processor_worker, daemon=True)
-worker_thread.start()
-
-def save_task_state(task):
-    """Guarda el estado de la tarea en un archivo JSON"""
-    task_file = os.path.join(PERSIST_DIR, f"{task.task_id}.json")
-    with open(task_file, 'w') as f:
-        json.dump(task.to_dict(), f)
-
-def load_task_state(task_id):
-    """Carga el estado de una tarea desde archivo"""
-    task_file = os.path.join(PERSIST_DIR, f"{task_id}.json")
-    if os.path.exists(task_file):
-        with open(task_file, 'r') as f:
-            return json.load(f)
-    return None
-
-def process_video_task(task):
-    """Procesa una tarea de video paso a paso"""
+    if not vids:
+        raise RuntimeError("Sin vídeos disponibles")
+
+    segs, acc = [], 0
+    for path in vids:
+        if acc >= adur + 2:
+            break
+        clip = VideoFileClip(path)
+        seg = clip.subclip(0, min(8, clip.duration))
+        segs.append(seg)
+        acc += seg.duration
+    base = concatenate_videoclips(segs, method="chain")
+    if base.duration < adur:
+        loops = math.ceil(adur / base.duration)
+        base = concatenate_videoclips([base] * loops, method="chain")
+    base = base.subclip(0, adur)
+
+    if music_fp:
+        mclip = loop_audio(AudioFileClip(music_fp), adur).volumex(0.2)
+        audio = CompositeAudioClip([mclip, voice_clip])
+    else:
+        audio = voice_clip
+
+    subs = make_subs_clips(script, base.w, base.h, adur)
+    grain = make_grain_clip((base.w, base.h), adur)
+    final_vid = CompositeVideoClip([base, grain, *subs]).set_audio(audio)
+
+    out_path = os.path.join(tmp, "final.mp4")
+    final_vid.write_videofile(out_path, fps=24, codec="libx264", audio_codec="aac", logger=None)
+    return out_path
+
+# ───────── async tasks ────────────────────────────────────────────────────────────
+def worker(tid: str, mode: str, topic: str, user_script: str, music: str | None):
     try:
-        task.status = "processing"
-        task.progress = 0
-        save_task_state(task)
-        
-        # Paso 1: Generar guión
-        task.progress = 10
-        save_task_state(task)
-        
-        if task.prompt_type == "Generar Guion con IA":
-            guion = generate_script_simple(task.input_text)
-        else:
-            guion = task.input_text.strip()
-            
-        task.steps_completed.append("guion_generado")
-        task.progress = 20
-        save_task_state(task)
-        
-        # Paso 2: Generar audio TTS
-        audio_path = os.path.join(PERSIST_DIR, f"{task.task_id}_audio.mp3")
-        success = asyncio.run(text_to_speech_simple(guion, audio_path))
-        
-        if not success:
-            raise Exception("Error generando audio")
-            
-        task.steps_completed.append("audio_generado")
-        task.progress = 40
-        save_task_state(task)
-        
-        # Paso 3: Buscar videos (simplificado)
-        keywords = extract_keywords_simple(guion)
-        video_urls = search_videos_simple(keywords)
-        
-        task.steps_completed.append("videos_encontrados")
-        task.progress = 60
-        save_task_state(task)
-        
-        # Paso 4: Crear video final (simplificado)
-        output_path = os.path.join(PERSIST_DIR, f"{task.task_id}_final.mp4")
-        
-        # Simulación de creación de video
-        # En producción, aquí iría tu lógica de moviepy
-        create_simple_video(video_urls, audio_path, output_path, task.musica_file)
-        
-        task.steps_completed.append("video_creado")
-        task.progress = 100
-        task.status = "completed"
-        task.result = output_path
-        save_task_state(task)
-        
+        txt = topic if mode == "Generar Guion con IA" else user_script
+        res_tmp = build_video(txt, mode == "Generar Guion con IA", music)
+        final_path = os.path.join(RESULTS_DIR, f"{tid}.mp4")
+        shutil.copy2(res_tmp, final_path)
+        TASKS[tid] = {"status": "done", "result": final_path, "ts": datetime.utcnow()}
     except Exception as e:
-        logger.error(f"Error procesando tarea {task.task_id}: {str(e)}")
-        task.status = "error"
-        task.error = str(e)
-        save_task_state(task)
-
-# Funciones simplificadas
-def generate_script_simple(prompt):
-    """Versión simplificada de generación de guión"""
-    # Aquí puedes usar tu lógica GPT-2 existente
-    return f"Este es un video sobre {prompt}. Es fascinante y educativo."
-
-async def text_to_speech_simple(text, output_path):
-    """TTS simplificado"""
-    try:
-        communicate = edge_tts.Communicate(text, "es-ES-JuanNeural")
-        await communicate.save(output_path)
-        return os.path.exists(output_path)
-    except:
-        return False
-
-def extract_keywords_simple(text):
-    """Extracción simple de keywords"""
-    words = text.lower().split()
-    # Filtrar palabras comunes
-    keywords = [w for w in words if len(w) > 4][:3]
-    return keywords if keywords else ["nature", "video", "background"]
-
-def search_videos_simple(keywords):
-    """Búsqueda simplificada de videos"""
-    # Aquí iría tu lógica de Pexels
-    # Por ahora retornamos URLs de ejemplo
-    return ["video1.mp4", "video2.mp4"]
-
-def create_simple_video(video_urls, audio_path, output_path, music_path=None):
-    """Creación simplificada de video"""
-    # Aquí iría tu lógica de MoviePy
-    # Por ahora creamos un archivo dummy
-    with open(output_path, 'w') as f:
-        f.write("dummy video content")
-    time.sleep(2)  # Simular procesamiento
-
-# Interfaz Gradio mejorada
-def submit_video_request(prompt_type, prompt_ia, prompt_manual, musica_file):
-    """Envía una solicitud de video a la cola"""
-    input_text = prompt_ia if prompt_type == "Generar Guion con IA" else prompt_manual
-    
-    if not input_text or not input_text.strip():
-        return None, "Por favor ingresa un texto"
-    
-    task_id = str(uuid.uuid4())
-    task = VideoTask(task_id, prompt_type, input_text, musica_file)
-    
-    # Guardar estado inicial
-    task_status[task_id] = task
-    save_task_state(task)
-    
-    # Añadir a la cola
-    processing_queue.put(task)
-    
-    return task_id, f"Tarea creada: {task_id}"
-
-def check_video_status(task_id):
-    """Verifica el estado de una tarea"""
-    if not task_id:
-        return "No hay ID de tarea", None, None
-    
-    # Intentar cargar desde archivo
-    task_data = load_task_state(task_id)
-    
-    if not task_data:
-        return "Tarea no encontrada", None, None
-    
-    status = task_data['status']
-    progress = task_data['progress']
-    
-    if status == "pending":
-        return f"⏳ En cola... ({progress}%)", None, None
-    elif status == "processing":
-        steps = ", ".join(task_data['steps_completed'])
-        return f"🔄 Procesando... ({progress}%) - Completado: {steps}", None, None
-    elif status == "completed":
-        video_path = task_data['result']
-        if os.path.exists(video_path):
-            return "✅ Video completado!", video_path, video_path
-        else:
-            return "❌ Video completado pero archivo no encontrado", None, None
-    elif status == "error":
-        return f"❌ Error: {task_data['error']}", None, None
-    
-    return "Estado desconocido", None, None
-
-# Interfaz Gradio
-with gr.Blocks(title="Generador de Videos con IA") as app:
-    gr.Markdown("# 🎬 Generador de Videos con IA (Sistema de Cola)")
-    
+        TASKS[tid] = {"status": "error", "error": str(e), "ts": datetime.utcnow()}
+
+def submit(mode, topic, user_script, music):
+    content = topic if mode == "Generar Guion con IA" else user_script
+    if not content.strip():
+        return "", "Ingresa texto"
+    tid = uuid.uuid4().hex[:8]
+    TASKS[tid] = {"status": "processing", "ts": datetime.utcnow()}
+    threading.Thread(target=worker, args=(tid, mode, topic, user_script, music), daemon=True).start()
+    return tid, f"Tarea {tid} creada"
+
+def check(tid):
+    if tid not in TASKS:
+        return None, None, "ID inválido"
+    info = TASKS[tid]
+    stat = info["status"]
+    if stat == "processing":
+        return None, None, "Procesando..."
+    if stat == "error":
+        return None, None, f"Error: {info['error']}"
+    return info["result"], info["result"], "Vídeo listo 🎉"
+
+# ───────── janitor thread ─────────────────────────────────────────────────────────
+def janitor():
+    while True:
+        now = datetime.utcnow()
+        for fname in os.listdir(RESULTS_DIR):
+            fpath = os.path.join(RESULTS_DIR, fname)
+            try:
+                mtime = datetime.utcfromtimestamp(os.path.getmtime(fpath))
+                if now - mtime > timedelta(hours=24):
+                    os.remove(fpath)
+                    for k, v in list(TASKS.items()):
+                        if v.get("result") == fpath:
+                            del TASKS[k]
+            except Exception:
+                pass
+        time.sleep(3600)
+
+threading.Thread(target=janitor, daemon=True).start()
+
+# ───────── gradio ui ─────────────────────────────────────────────────────────────
+with gr.Blocks(title="Generador de Vídeos IA") as demo:
     with gr.Tabs():
-        with gr.TabItem("Crear Video"):
-            with gr.Row():
-                with gr.Column():
-                    prompt_type = gr.Radio(
-                        ["Generar Guion con IA", "Usar Mi Guion"],
-                        label="Método de Entrada",
-                        value="Generar Guion con IA"
-                    )
-                    
-                    prompt_ia = gr.Textbox(
-                        label="Tema para IA",
-                        placeholder="Describe el tema del video..."
-                    )
-                    
-                    prompt_manual = gr.Textbox(
-                        label="Tu Guion Completo",
-                        placeholder="Escribe tu guion aquí...",
-                        visible=False
-                    )
-                    
-                    musica_input = gr.Audio(
-                        label="Música de fondo (opcional)",
-                        type="filepath"
-                    )
-                    
-                    submit_btn = gr.Button("📤 Enviar a Cola", variant="primary")
-                
-                with gr.Column():
-                    task_id_output = gr.Textbox(
-                        label="ID de Tarea",
-                        interactive=False
-                    )
-                    submit_status = gr.Textbox(
-                        label="Estado de Envío",
-                        interactive=False
-                    )
-        
-        with gr.TabItem("Verificar Estado"):
-            with gr.Row():
-                with gr.Column():
-                    task_id_input = gr.Textbox(
-                        label="ID de Tarea",
-                        placeholder="Pega aquí el ID de tu tarea..."
-                    )
-                    check_btn = gr.Button("🔍 Verificar Estado")
-                    auto_check = gr.Checkbox(
-                        label="Verificar automáticamente cada 5 segundos"
-                    )
-                
-                with gr.Column():
-                    status_output = gr.Textbox(
-                        label="Estado Actual",
-                        interactive=False
-                    )
-                    video_output = gr.Video(
-                        label="Video Generado",
-                        interactive=False
-                    )
-                    download_output = gr.File(
-                        label="Descargar Video",
-                        interactive=False
-                    )
-    
-    # Eventos
-    prompt_type.change(
-        lambda x: (
-            gr.update(visible=x == "Generar Guion con IA"),
-            gr.update(visible=x == "Usar Mi Guion")
-        ),
-        inputs=prompt_type,
-        outputs=[prompt_ia, prompt_manual]
-    )
-    
-    submit_btn.click(
-        submit_video_request,
-        inputs=[prompt_type, prompt_ia, prompt_manual, musica_input],
-        outputs=[task_id_output, submit_status]
-    )
-    
-    check_btn.click(
-        check_video_status,
-        inputs=[task_id_input],
-        outputs=[status_output, video_output, download_output]
-    )
-    
-    # Auto-check cada 5 segundos si está activado
-    def auto_check_status(task_id, should_check):
-        if should_check and task_id:
-            return check_video_status(task_id)
-        return gr.update(), gr.update(), gr.update()
-    
-    auto_check.change(
-        lambda x: gr.update(visible=x),
-        inputs=[auto_check],
-        outputs=[check_btn]
+        with gr.TabItem("Crear Vídeo"):
+            mode = gr.Radio(["Generar Guion con IA", "Usar Mi Guion"], value="Generar Guion con IA")
+            topic = gr.Textbox(label="Tema")
+            user_script = gr.Textbox(label="Guion Completo", visible=False)
+            music = gr.Audio(type="filepath", label="Música (opcional)")
+            btn = gr.Button("Generar")
+            tid_out = gr.Textbox(label="ID de tarea")
+            msg = gr.Textbox(label="Estado")
+        with gr.TabItem("Revisar Estado"):
+            tid_in = gr.Textbox(label="ID de tarea")
+            chk = gr.Button("Verificar")
+            vid = gr.Video()
+            dlf = gr.File()
+
+    mode.change(
+        lambda m: (gr.update(visible=m == "Generar Guion con IA"), gr.update(visible=m != "Generar Guion con IA")),
+        mode, [topic, user_script]
     )
+    btn.click(submit, [mode, topic, user_script, music], [tid_out, msg])
+    chk.click(check, tid_in, [vid, dlf, msg])
 
 if __name__ == "__main__":
-    app.launch(server_name="0.0.0.0", server_port=7860)
+    demo.launch()