character_detection.py

"""

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 jerárquico aglomerativo

4. Generación de carpetas por personaje

"""
import cv2
import os
import json
import logging
import shutil
from pathlib import Path
import numpy as np
from scipy.cluster.hierarchy import linkage, fcluster
from collections import Counter
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, *, start_offset_sec: float = 3.0, extract_every_sec: float = 0.5,

                                 detector_backend: str = 'retinaface', min_face_area: int = 100,

                                 enforce_detection: bool = False) -> 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 = float(extract_every_sec)
        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
        start_frame = int(max(0.0, start_offset_sec) * (fps if fps > 0 else 25))
        
        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 < start_frame:
                frame_count += 1
                continue

            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=detector_backend,
                        enforce_detection=enforce_detection
                    )
                    
                    if face_objs:
                        for i, face_obj in enumerate(face_objs):
                            embedding = face_obj['embedding']
                            facial_area = face_obj.get('facial_area', {})
                            try:
                                w = int(facial_area.get('w', 0))
                                h = int(facial_area.get('h', 0))
                                if w * h < int(min_face_area):
                                    continue
                            except Exception:
                                pass
                            
                            # Guardar recorte de la cara (mejor para UI y clustering visual)
                            x = int(facial_area.get('x', 0)); y = int(facial_area.get('y', 0))
                            w = int(facial_area.get('w', 0)); h = int(facial_area.get('h', 0))
                            x2 = max(0, x); y2 = max(0, y)
                            x3 = min(frame.shape[1], x + w); y3 = min(frame.shape[0], y + h)
                            crop = frame[y2:y3, x2:x3] if (x3 > x2 and y3 > y2) else frame
                            save_path = self.faces_dir / f"face_{saved_count:04d}.jpg"
                            cv2.imwrite(str(save_path), crop)
                            
                            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], max_groups: int, min_samples: int) -> np.ndarray:
        """

        Agrupa caras similares usando clustering jerárquico aglomerativo con selección óptima.

        Selecciona automáticamente el mejor número de clusters usando silhouette score.

        

        Args:

            embeddings_caras: Lista de embeddings de caras

            max_groups: Número máximo de clusters a formar

            min_samples: Tamaño mínimo de cluster válido

        

        Returns:

            Array de labels (cluster asignado a cada cara, -1 para ruido)

        """
        if not embeddings_caras:
            return np.array([])
        
        logger.info(f"Clustering {len(embeddings_caras)} caras con max_groups={max_groups}, min_samples={min_samples}")
        
        # Extraer solo los embeddings
        X = np.array([cara['embeddings'] for cara in embeddings_caras])
        
        if len(X) < min_samples:
            # Si hay menos muestras que el mínimo, todo es ruido
            return np.full(len(X), -1, dtype=int)
        
        # Linkage usando average linkage (más flexible que ward, menos sensible a outliers)
        # Esto ayuda a agrupar mejor la misma persona con diferentes ángulos/expresiones
        Z = linkage(X, method='average', metric='cosine')  # Cosine similarity para embeddings
        
        # Encontrar el número óptimo de clusters usando silhouette score
        from sklearn.metrics import silhouette_score
        best_n_clusters = 2
        best_score = -1
        
        max_to_try = min(max_groups, len(X) - 1)
        
        if max_to_try >= 2:
            for n_clusters in range(2, max_to_try + 1):
                trial_labels = fcluster(Z, t=n_clusters, criterion='maxclust') - 1
                
                trial_counts = Counter(trial_labels)
                valid_clusters = sum(1 for count in trial_counts.values() if count >= min_samples)
                
                if valid_clusters >= 2:
                    try:
                        score = silhouette_score(X, trial_labels, metric='cosine')
                        # Penalización MUY fuerte para reducir duplicados de la misma persona
                        # Valores: 0.05 = fuerte, 0.07 = muy fuerte, 0.10 = extremo
                        adjusted_score = score - (n_clusters * 0.07)
                        
                        if adjusted_score > best_score:
                            best_score = adjusted_score
                            best_n_clusters = n_clusters
                    except:
                        pass
        
        logger.info(f"Clustering óptimo: {best_n_clusters} clusters (de máximo {max_groups}), silhouette: {best_score:.3f}")
        labels = fcluster(Z, t=best_n_clusters, criterion='maxclust') - 1
        
        # Filtrar clusters pequeños
        label_counts = Counter(labels)
        filtered_labels = []
        for lbl in labels:
            if label_counts[lbl] >= min_samples:
                filtered_labels.append(lbl)
            else:
                filtered_labels.append(-1)
        labels = np.array(filtered_labels, dtype=int)
        
        # 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 válidos 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, valida caras y guarda metadata.

        Integra validación con DeepFace para filtrar falsos positivos y detectar género.

        

        Args:

            embeddings_caras: Lista de embeddings de caras

            labels: Array de labels de clustering

        

        Returns:

            Lista de personajes detectados con metadata (solo clusters válidos)

        """
        from face_classifier import validate_and_classify_face, FACE_CONFIDENCE_THRESHOLD
        
        characters_validated = []
        
        # 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"Procesando {len(clusters)} clusters detectados...")
        original_cluster_count = len(clusters)
        
        # Procesar cada cluster
        for cluster_id, face_indices in clusters.items():
            char_id = f"char_{cluster_id:02d}"
            
            # PASO 1: Ordenar caras por score (usar área como proxy de calidad)
            # Caras más grandes = mejor detección
            face_detections = []
            for face_idx in face_indices:
                face_data = embeddings_caras[face_idx]
                facial_area = face_data.get('facial_area', {})
                w = facial_area.get('w', 0)
                h = facial_area.get('h', 0)
                area_score = w * h  # Score basado en área
                
                face_detections.append({
                    'index': face_idx,
                    'score': area_score,
                    'facial_area': facial_area,
                    'path': face_data['path']
                })
            
            # Ordenar por score descendente (mejores primero)
            face_detections_sorted = sorted(
                face_detections,
                key=lambda x: x['score'],
                reverse=True
            )
            
            if not face_detections_sorted:
                logger.info(f"[VALIDATION] ✗ Cluster {char_id}: sense deteccions, eliminant")
                continue
            
            # PASO 2: Validar SOLO la mejor cara del cluster
            best_face = face_detections_sorted[0]
            best_face_path = best_face['path']
            
            logger.info(f"[VALIDATION] Cluster {char_id}: validant millor cara (score={best_face['score']:.0f}px²)")
            
            validation = validate_and_classify_face(best_face_path)
            
            if not validation:
                logger.info(f"[VALIDATION] ✗ Cluster {char_id}: error en validació, eliminant")
                continue
            
            # PASO 3: Verificar si és una cara vàlida
            if not validation['is_valid_face'] or validation['face_confidence'] < FACE_CONFIDENCE_THRESHOLD:
                logger.info(f"[VALIDATION] ✗ Cluster {char_id}: score baix ({validation['face_confidence']:.2f}), eliminant tot el clúster")
                continue
            
            # PASO 4: És una cara vàlida! Crear carpeta
            char_dir = self.output_base / char_id
            char_dir.mkdir(parents=True, exist_ok=True)
            
            # PASO 5: Limitar caras a mostrar (primera meitat + 1)
            total_faces = len(face_detections_sorted)
            max_faces_to_show = (total_faces // 2) + 1
            face_detections_limited = face_detections_sorted[:max_faces_to_show]
            
            # Copiar solo las caras limitadas
            face_files = []
            for i, face_det in enumerate(face_detections_limited):
                src_path = Path(face_det['path'])
                dst_path = char_dir / f"face_{i:03d}.jpg"
                if src_path.exists():
                    shutil.copy(src_path, dst_path)
                    face_files.append(f"/files/{self.video_name}/{char_id}/face_{i:03d}.jpg")
            
            # Imagen representativa (la mejor)
            representative_src = Path(best_face_path)
            representative_dst = char_dir / "representative.jpg"
            if representative_src.exists():
                shutil.copy(representative_src, representative_dst)
            
            # PASO 6: Generar nombre de clúster
            cluster_number = int(char_id.split('_')[1]) + 1
            character_name = f"Cluster {cluster_number}"
            gender = validation['gender']
            
            # Metadata del personaje
            image_url = f"/files/{self.video_name}/{char_id}/representative.jpg"
            
            character_data = {
                "id": char_id,
                "name": character_name,
                "gender": gender,
                "gender_confidence": validation['gender_confidence'],
                "face_confidence": validation['face_confidence'],
                "man_prob": validation['man_prob'],
                "woman_prob": validation['woman_prob'],
                "image_path": str(representative_dst),
                "image_url": image_url,
                "face_files": face_files,
                "num_faces": len(face_detections_limited),
                "total_faces_detected": total_faces,
                "folder": str(char_dir)
            }
            
            characters_validated.append(character_data)
            
            logger.info(f"[VALIDATION] ✓ Cluster {char_id}: cara vàlida! "
                       f"Nom={character_name}, Gender={gender} (conf={validation['gender_confidence']:.2f}), "
                       f"Mostrant {len(face_detections_limited)}/{total_faces} cares")
        
        # Estadístiques finals
        eliminated_count = original_cluster_count - len(characters_validated)
        logger.info(f"[VALIDATION] Total: {len(characters_validated)} clústers vàlids "
                   f"(eliminats {eliminated_count} falsos positius)")
        
        return characters_validated
    
    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, max_groups: int = 3, min_cluster_size: int = 3,

                          *, start_offset_sec: float = 3.0, extract_every_sec: float = 0.5) -> Tuple[List[Dict], Path, np.ndarray, List[Dict[str, Any]]]:
        """

        Pipeline completo de detección de personajes con clustering jerárquico.

        

        Args:

            max_groups: Número máximo de clusters a formar

            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(start_offset_sec=start_offset_sec, extract_every_sec=extract_every_sec)
        
        # 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, max_groups, min_cluster_size)
        
        # 6. Crear carpetas de personajes
        characters = self.create_character_folders(embeddings_caras, labels)
        
        return characters, analysis_path, labels, embeddings_caras


# Función de conveniencia para usar en el API
def detect_characters_from_video(video_path: str, output_base: str, 

                                 max_groups: int = 3, min_cluster_size: int = 3,

                                 video_name: str = None,

                                 *, start_offset_sec: float = 3.0, extract_every_sec: float = 0.5) -> Dict[str, Any]:
    """

    Función de alto nivel para detectar personajes en un vídeo usando clustering jerárquico.

    

    Args:

        video_path: Ruta al vídeo

        output_base: Directorio base para guardar resultados

        max_groups: Número máximo de clusters a formar

        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, labels, embeddings_caras = detector.detect_characters(max_groups, min_cluster_size,
                                                                                    start_offset_sec=start_offset_sec,
                                                                                    extract_every_sec=extract_every_sec)
    
    return {
        "characters": characters,
        "analysis_path": str(analysis_path),
        "num_characters": len(characters),
        "face_labels": labels.tolist() if isinstance(labels, np.ndarray) else list(labels),
        "num_face_embeddings": len(embeddings_caras)
    }