from __future__ import annotations from fastapi import FastAPI, UploadFile, File, Form, BackgroundTasks, HTTPException from fastapi import Body from fastapi.responses import JSONResponse, FileResponse from fastapi.middleware.cors import CORSMiddleware from pathlib import Path import shutil import uvicorn import json import uuid from datetime import datetime from typing import Dict from enum import Enum import os from video_processing import process_video_pipeline from audio_tools import process_audio_for_video, extract_audio_ffmpeg, embed_voice_segments from casting_loader import ensure_chroma, build_faces_index, build_voices_index from narration_system import NarrationSystem from llm_router import load_yaml, LLMRouter from character_detection import detect_characters_from_video from pipelines.audiodescription import generate as ad_generate app = FastAPI(title="Veureu Engine API", version="0.2.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) ROOT = Path("/tmp/veureu") ROOT.mkdir(parents=True, exist_ok=True) TEMP_ROOT = Path("/tmp/temp") TEMP_ROOT.mkdir(parents=True, exist_ok=True) VIDEOS_ROOT = Path("/tmp/data/videos") VIDEOS_ROOT.mkdir(parents=True, exist_ok=True) IDENTITIES_ROOT = Path("/tmp/characters") IDENTITIES_ROOT.mkdir(parents=True, exist_ok=True) # Sistema de jobs asíncronos class JobStatus(str, Enum): QUEUED = "queued" PROCESSING = "processing" DONE = "done" FAILED = "failed" jobs: Dict[str, dict] = {} def hierarchical_cluster_with_min_size(X, max_groups: int, min_cluster_size: int): """ Clustering jerárquico aglomerativo que produce hasta max_groups clusters. Filtra clusters con menos de min_cluster_size muestras (marcados como -1/ruido). Args: X: Array de embeddings (N, D) max_groups: Número máximo de clusters a formar min_cluster_size: Tamaño mínimo de cluster válido Returns: Array de labels (N,) donde -1 indica ruido """ import numpy as np from scipy.cluster.hierarchy import linkage, fcluster from collections import Counter if len(X) == 0: return np.array([]) if len(X) < min_cluster_size: # Si hay menos muestras que el mínimo, todo es ruido return np.full(len(X), -1, dtype=int) # Linkage usando distancia euclidiana con método 'ward' Z = linkage(X, method='ward', metric='euclidean') # Cortar el dendrograma en max_groups clusters labels = fcluster(Z, t=max_groups, criterion='maxclust') # fcluster devuelve labels 1-indexed, convertir a 0-indexed labels = labels - 1 # Filtrar clusters pequeños label_counts = Counter(labels) filtered_labels = [] for lbl in labels: if label_counts[lbl] >= min_cluster_size: filtered_labels.append(lbl) else: filtered_labels.append(-1) # Ruido return np.array(filtered_labels, dtype=int) @app.get("/") def root(): return {"ok": True, "service": "veureu-engine"} @app.post("/process_video") async def process_video( video_file: UploadFile = File(...), config_path: str = Form("config.yaml"), out_root: str = Form("results"), db_dir: str = Form("chroma_db"), ): tmp_video = ROOT / video_file.filename with tmp_video.open("wb") as f: shutil.copyfileobj(video_file.file, f) result = process_video_pipeline(str(tmp_video), config_path=config_path, out_root=out_root, db_dir=db_dir) return JSONResponse(result) @app.post("/create_initial_casting") async def create_initial_casting( background_tasks: BackgroundTasks, video: UploadFile = File(...), max_groups: int = Form(5), min_cluster_size: int = Form(3), voice_max_groups: int = Form(5), voice_min_cluster_size: int = Form(3), max_frames: int = Form(100), ): """ Crea un job para procesar el vídeo de forma asíncrona usando clustering jerárquico. Devuelve un job_id inmediatamente. """ # Guardar vídeo en carpeta de datos video_name = Path(video.filename).stem dst_video = VIDEOS_ROOT / f"{video_name}.mp4" with dst_video.open("wb") as f: shutil.copyfileobj(video.file, f) # Crear job_id único job_id = str(uuid.uuid4()) # Inicializar el job jobs[job_id] = { "id": job_id, "status": JobStatus.QUEUED, "video_path": str(dst_video), "video_name": video_name, "max_groups": int(max_groups), "min_cluster_size": int(min_cluster_size), "voice_max_groups": int(voice_max_groups), "voice_min_cluster_size": int(voice_min_cluster_size), "max_frames": int(max_frames), "created_at": datetime.now().isoformat(), "results": None, "error": None } print(f"[{job_id}] Job creado para vídeo: {video_name}") # Iniciar procesamiento en background background_tasks.add_task(process_video_job, job_id) # Devolver job_id inmediatamente return {"job_id": job_id} @app.get("/jobs/{job_id}/status") def get_job_status(job_id: str): """ Devuelve el estado actual de un job. El UI hace polling de este endpoint cada 5 segundos. """ if job_id not in jobs: raise HTTPException(status_code=404, detail="Job not found") job = jobs[job_id] # Normalizar el estado a string status_value = job["status"].value if isinstance(job["status"], JobStatus) else str(job["status"]) response = {"status": status_value} # Incluir resultados si existen (evita condiciones de carrera) if job.get("results") is not None: response["results"] = job["results"] # Incluir error si existe if job.get("error"): response["error"] = job["error"] return response @app.get("/files/{video_name}/{char_id}/{filename}") def serve_character_file(video_name: str, char_id: str, filename: str): """ Sirve archivos estáticos de personajes (imágenes). Ejemplo: /files/dif_catala_1/char1/representative.jpg """ # Las caras se guardan en /tmp/temp/