""" Character Detection Module Integra el trabajo de Ana para detección de personajes mediante: 1. Extracción de caras y embeddings 2. Extracción de voces y embeddings 3. Clustering con DBSCAN 4. Generación de carpetas por personaje """ import cv2 import os import json import logging import shutil from pathlib import Path from sklearn.cluster import DBSCAN import numpy as np from typing import List, Dict, Any, Tuple # Imports de las herramientas de vision y audio desde los módulos de la raíz try: # DeepFace para detección y embeddings de caras from deepface import DeepFace DEEPFACE_AVAILABLE = True except Exception as e: DEEPFACE_AVAILABLE = False logging.warning(f"DeepFace no disponible: {e}") logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class CharacterDetector: """ Detector de personajes que integra el trabajo de Ana. """ def __init__(self, video_path: str, output_base: Path, video_name: str = None): """ Args: video_path: Ruta al archivo de vídeo output_base: Directorio base para guardar resultados (ej: /tmp/temp/video_name) video_name: Nombre del vídeo (para construir URLs) """ self.video_path = video_path self.output_base = Path(output_base) self.output_base.mkdir(parents=True, exist_ok=True) self.video_name = video_name or self.output_base.name # Crear subdirectorios self.faces_dir = self.output_base / "faces" self.voices_dir = self.output_base / "voices" self.scenes_dir = self.output_base / "scenes" for d in [self.faces_dir, self.voices_dir, self.scenes_dir]: d.mkdir(parents=True, exist_ok=True) def extract_faces_embeddings(self) -> List[Dict[str, Any]]: """ Extrae caras del vídeo y calcula sus embeddings usando DeepFace directamente. Returns: Lista de dicts con {"embeddings": [...], "path": "..."} """ if not DEEPFACE_AVAILABLE: logger.warning("DeepFace no disponible, retornando lista vacía") return [] logger.info("Extrayendo caras del vídeo con DeepFace...") extract_every = 1.0 # segundos video = cv2.VideoCapture(self.video_path) fps = int(video.get(cv2.CAP_PROP_FPS)) total_frames = int(video.get(cv2.CAP_PROP_FRAME_COUNT)) frame_interval = int(fps * extract_every) frame_count = 0 saved_count = 0 embeddings_caras = [] logger.info(f"Total frames: {total_frames}, FPS: {fps}, Procesando cada {frame_interval} frames") while True: ret, frame = video.read() if not ret: break if frame_count % frame_interval == 0: temp_path = self.faces_dir / "temp_frame.jpg" cv2.imwrite(str(temp_path), frame) try: # Extraer embeddings con DeepFace # represent() devuelve una lista de dicts, uno por cada cara detectada face_objs = DeepFace.represent( img_path=str(temp_path), model_name='Facenet512', detector_backend='opencv', enforce_detection=False ) if face_objs: for i, face_obj in enumerate(face_objs): embedding = face_obj['embedding'] facial_area = face_obj.get('facial_area', {}) # Guardar el frame completo save_path = self.faces_dir / f"frame_{saved_count:04d}.jpg" cv2.imwrite(str(save_path), frame) embeddings_caras.append({ "embeddings": embedding, "path": str(save_path), "frame": frame_count, "facial_area": facial_area }) saved_count += 1 if frame_count % (frame_interval * 10) == 0: logger.info(f"Progreso: frame {frame_count}/{total_frames}, caras detectadas: {saved_count}") except Exception as e: logger.debug(f"No se detectaron caras en frame {frame_count}: {e}") if temp_path.exists(): os.remove(temp_path) frame_count += 1 video.release() logger.info(f"✓ Caras extraídas: {len(embeddings_caras)}") return embeddings_caras def extract_voices_embeddings(self) -> List[Dict[str, Any]]: """ Extrae voces del vídeo y calcula sus embeddings. Por ahora retorna lista vacía (funcionalidad opcional). Returns: Lista de dicts con {"embeddings": [...], "path": "..."} """ logger.info("Extracción de voces deshabilitada temporalmente") return [] def extract_scenes_embeddings(self) -> List[Dict[str, Any]]: """ Extrae escenas clave del vídeo. Por ahora retorna lista vacía (funcionalidad opcional). Returns: Lista de dicts con {"embeddings": [...], "path": "..."} """ logger.info("Extracción de escenas deshabilitada temporalmente") return [] def cluster_faces(self, embeddings_caras: List[Dict], epsilon: float, min_samples: int) -> np.ndarray: """ Agrupa caras similares usando DBSCAN. Basado en get_face_clusters de Ana. Args: embeddings_caras: Lista de embeddings de caras epsilon: Parámetro eps de DBSCAN min_samples: Parámetro min_samples de DBSCAN Returns: Array de labels (cluster asignado a cada cara) """ if not embeddings_caras: return np.array([]) logger.info(f"Clustering {len(embeddings_caras)} caras con eps={epsilon}, min_samples={min_samples}") # Extraer solo los embeddings X = np.array([cara['embeddings'] for cara in embeddings_caras]) # DBSCAN clustering clustering = DBSCAN(eps=epsilon, min_samples=min_samples, metric='euclidean').fit(X) labels = clustering.labels_ # Contar clusters (excluyendo ruido -1) n_clusters = len(set(labels)) - (1 if -1 in labels else 0) n_noise = list(labels).count(-1) logger.info(f"Clusters encontrados: {n_clusters}, Ruido: {n_noise}") return labels def create_character_folders(self, embeddings_caras: List[Dict], labels: np.ndarray) -> List[Dict[str, Any]]: """ Crea carpetas para cada personaje detectado y guarda las caras. Args: embeddings_caras: Lista de embeddings de caras labels: Array de labels de clustering Returns: Lista de personajes detectados con metadata """ characters = [] # Agrupar caras por cluster clusters = {} for idx, label in enumerate(labels): if label == -1: # Ignorar ruido continue if label not in clusters: clusters[label] = [] clusters[label].append(idx) logger.info(f"Creando carpetas para {len(clusters)} personajes...") # Crear carpeta para cada personaje for cluster_id, face_indices in clusters.items(): char_id = f"char{cluster_id + 1}" char_dir = self.output_base / char_id char_dir.mkdir(parents=True, exist_ok=True) # Copiar todas las caras del personaje for i, face_idx in enumerate(face_indices): src_path = Path(embeddings_caras[face_idx]['path']) dst_path = char_dir / f"face_{i:03d}.jpg" if src_path.exists(): shutil.copy(src_path, dst_path) # Seleccionar imagen representativa (primera cara) if face_indices: representative_src = Path(embeddings_caras[face_indices[0]]['path']) representative_dst = char_dir / "representative.jpg" if representative_src.exists(): shutil.copy(representative_src, representative_dst) # Metadata del personaje # Construir URL relativa para la imagen image_url = f"/files/{self.video_name}/{char_id}/representative.jpg" characters.append({ "id": char_id, "name": f"Personatge {cluster_id + 1}", "image_path": str(char_dir / "representative.jpg"), # Ruta local "image_url": image_url, # URL para el API "num_faces": len(face_indices), "folder": str(char_dir) }) logger.info(f"Carpetas creadas para {len(characters)} personajes") return characters def save_analysis_json(self, embeddings_caras: List[Dict], embeddings_voices: List[Dict], embeddings_escenas: List[Dict]) -> Path: """ Guarda el análisis completo en un archivo JSON. Similar al analysis.json de Ana. Returns: Path al archivo JSON guardado """ analysis_data = { "caras": embeddings_caras, "voices": embeddings_voices, "escenas": embeddings_escenas } analysis_path = self.output_base / "analysis.json" try: with open(analysis_path, "w", encoding="utf-8") as f: json.dump(analysis_data, f, indent=2, ensure_ascii=False) logger.info(f"Analysis JSON guardado: {analysis_path}") except Exception as e: logger.warning(f"Error al guardar analysis JSON: {e}") return analysis_path def detect_characters(self, epsilon: float = 0.5, min_cluster_size: int = 2) -> Tuple[List[Dict], Path]: """ Pipeline completo de detección de personajes. Args: epsilon: Parámetro epsilon para DBSCAN min_cluster_size: Tamaño mínimo de cluster Returns: Tuple de (lista de personajes, path al analysis.json) """ # 1. Extraer caras y embeddings embeddings_caras = self.extract_faces_embeddings() # 2. Extraer voces y embeddings (opcional, por ahora) embeddings_voices = self.extract_voices_embeddings() # 3. Extraer escenas y embeddings (opcional, por ahora) embeddings_escenas = self.extract_scenes_embeddings() # 4. Guardar análisis completo analysis_path = self.save_analysis_json(embeddings_caras, embeddings_voices, embeddings_escenas) # 5. Clustering de caras labels = self.cluster_faces(embeddings_caras, epsilon, min_cluster_size) # 6. Crear carpetas de personajes characters = self.create_character_folders(embeddings_caras, labels) return characters, analysis_path # Función de conveniencia para usar en el API def detect_characters_from_video(video_path: str, output_base: str, epsilon: float = 0.5, min_cluster_size: int = 2, video_name: str = None) -> Dict[str, Any]: """ Función de alto nivel para detectar personajes en un vídeo. Args: video_path: Ruta al vídeo output_base: Directorio base para guardar resultados epsilon: Parámetro epsilon para DBSCAN min_cluster_size: Tamaño mínimo de cluster video_name: Nombre del vídeo (para construir URLs) Returns: Dict con resultados: {"characters": [...], "analysis_path": "..."} """ detector = CharacterDetector(video_path, Path(output_base), video_name=video_name) characters, analysis_path = detector.detect_characters(epsilon, min_cluster_size) return { "characters": characters, "analysis_path": str(analysis_path), "num_characters": len(characters) }