Blame for processing/topic_detection.py

"""
topic_detection.py — Detección de temas via HuggingFace zero-shot classification.

Usa el modelo joeddav/xlm-roberta-large-xnli para clasificar textos contra
una lista fija de temas candidatos (TOPIC_CANDIDATES en config.py).
No requiere entrenamiento previo — funciona con etiquetas en lenguaje natural.

Resultado por actor: lista de dicts [{"tema": str, "score": float, "peso_relativo": float}]
Los resultados se cachean en la tabla Supabase actor_topics para no
recalcular en cada carga del dashboard.
"""

import logging
import random
import time
from datetime import datetime, timezone

import requests
from supabase import create_client

import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))

from config import (
    HUGGINGFACE_API_KEY,
    HF_API_URL_TOPICS,
    HF_RETRY_DELAY,
    HF_RATELIMIT_DELAY,
    SCRAPING_MAX_RETRIES,
    TOPIC_CANDIDATES,
    SUPABASE_URL,
    SUPABASE_KEY,
    SUPABASE_SERVICE_KEY,
    LOG_FORMAT,
    LOG_DATE_FORMAT,
    LOG_LEVEL,
)
from processing.sanitizer import limpiar_texto_para_nlp

logging.basicConfig(format=LOG_FORMAT, datefmt=LOG_DATE_FORMAT)
logger = logging.getLogger("topic_detection")
logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO))


def _headers() -> dict:
    """Construye los headers HTTP para la API de HuggingFace."""
    headers = {"Content-Type": "application/json"}
    if HUGGINGFACE_API_KEY:
        headers["Authorization"] = f"Bearer {HUGGINGFACE_API_KEY}"
    return headers


def _classify_text(texto: str) -> dict:
    """
    Clasifica un texto contra los TOPIC_CANDIDATES usando zero-shot classification.

    Devuelve dict con listas paralelas 'labels' y 'scores', o dict vacío si falla.
    """
    payload = {
        "inputs": texto,
        "parameters": {"candidate_labels": TOPIC_CANDIDATES},
        "options": {"wait_for_model": True},
    }

    for intento in range(1, SCRAPING_MAX_RETRIES + 1):
        try:
            resp = requests.post(
                HF_API_URL_TOPICS,
                headers=_headers(),
                json=payload,
                timeout=60,
            )

            if resp.status_code == 200:
                return resp.json()

            elif resp.status_code == 503:
                logger.warning(
                    f"[topic_detection] 503 modelo cargando (intento {intento}/"
                    f"{SCRAPING_MAX_RETRIES}) — esperando {HF_RETRY_DELAY}s"
                )
                time.sleep(HF_RETRY_DELAY)

            elif resp.status_code == 429:
                logger.warning(
                    f"[topic_detection] 429 rate limit — esperando {HF_RATELIMIT_DELAY}s"
                )
                time.sleep(HF_RATELIMIT_DELAY)

            else:
                logger.error(
                    f"[topic_detection] error {resp.status_code}: {resp.text[:200]}"
                )
                return {}

        except requests.exceptions.Timeout:
            logger.warning(
                f"[topic_detection] timeout en intento {intento}/{SCRAPING_MAX_RETRIES}"
            )
            if intento < SCRAPING_MAX_RETRIES:
                time.sleep(HF_RETRY_DELAY)
        except Exception as e:
            logger.error(f"[topic_detection] error inesperado: {e}")
            return {}

    return {}


MAX_RESUMEN_CHARS = 1800  # ~512 tokens en español; límite del modelo
SAMPLE_SIZE = 25          # posts a samplear; evita sesgo hacia los primeros N posts


def _construir_resumen_actor(posts: list[dict]) -> str:
    """
    Construye un texto representativo del actor muestreando aleatoriamente sus posts.

    Samplea hasta SAMPLE_SIZE posts al azar para que el análisis represente
    toda la actividad del actor, no solo los primeros que devuelve la query.
    Trunca a MAX_RESUMEN_CHARS para no superar el límite del modelo.
    """
    posts_con_texto = [
        p for p in posts
        if f"{p.get('titulo', '')}. {p.get('resumen', '')}".strip() not in ("", ".")
    ]

    if len(posts_con_texto) > SAMPLE_SIZE:
        posts_con_texto = random.sample(posts_con_texto, SAMPLE_SIZE)

    fragmentos = []
    total = 0
    for post in posts_con_texto:
        fragmento = f"{post.get('titulo', '')}. {post.get('resumen', '')}".strip()
        if total + len(fragmento) > MAX_RESUMEN_CHARS:
            restante = MAX_RESUMEN_CHARS - total
            if restante > 20:
                fragmentos.append(fragmento[:restante])
            break
        fragmentos.append(fragmento)
        total += len(fragmento)

    return " ".join(fragmentos)


def guardar_temas_supabase(actor_id: str, temas: list[dict]) -> None:
    """Guarda los temas detectados en actor_topics (upsert por actor_id + posicion)."""
    if not temas or not SUPABASE_URL or not SUPABASE_KEY:
        return
    try:
        client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY)
        ahora = datetime.now(timezone.utc).isoformat()
        rows = [
            {
                "actor_id": actor_id,
                "posicion": i + 1,
                "tema": t["tema"],
                "score": t["score"],
                "peso_relativo": t["peso_relativo"],
                "calculado_en": ahora,
            }
            for i, t in enumerate(temas)
        ]
        client.table("actor_topics").upsert(rows, on_conflict="actor_id,posicion").execute()
        logger.info(f"[topic_detection] {actor_id}: {len(rows)} temas guardados en actor_topics")
    except Exception as e:
        logger.warning(f"[topic_detection] error guardando temas en Supabase: {e}")


def get_actor_topics_from_db(actor_id: str) -> list[dict]:
    """
    Lee los temas cacheados desde actor_topics en Supabase.

    Devuelve lista vacía si no hay datos (actor sin pipeline ejecutado aún).
    """
    if not SUPABASE_URL or not SUPABASE_KEY:
        return []
    try:
        client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY)
        resp = (
            client.table("actor_topics")
            .select("tema, score, peso_relativo, calculado_en")
            .eq("actor_id", actor_id)
            .order("posicion")
            .execute()
        )
        return resp.data or []
    except Exception as e:
        logger.warning(f"[topic_detection] error leyendo actor_topics: {e}")
        return []


def get_actor_topics(actor_id: str, top_n: int = 5) -> list[dict]:
    """
    Detecta los principales temas de un actor usando UNA sola request a HuggingFace.

    Estrategia:
      - Samplea aleatoriamente hasta 25 posts (representatividad)
      - Hace 1 clasificación zero-shot sobre el resumen concatenado
      - Guarda resultados en actor_topics para no recalcular en el dashboard

    Parámetros:
        actor_id: ID del actor en seed_actors.json
        top_n:    cuántos temas devolver (default: 5)

    Devuelve:
        Lista de dicts [{"tema": str, "score": float, "peso_relativo": float}]
        ordenada de mayor a menor score. Lista vacía si no hay datos.
    """
    if not SUPABASE_URL or not SUPABASE_KEY:
        logger.warning("[topic_detection] Supabase no configurado")
        return []

    try:
        client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY)
        resp = (
            client.table("posts")
            .select("titulo, resumen")
            .eq("actor_id", actor_id)
            .limit(100)
            .execute()
        )
        posts = resp.data or []
    except Exception as e:
        logger.error(f"[topic_detection] error leyendo Supabase para {actor_id}: {e}")
        return []

    if not posts:
        logger.info(f"[topic_detection] {actor_id}: sin posts en Supabase")
        return []

    resumen = _construir_resumen_actor(posts)
    if not resumen:
        return []

    resumen = limpiar_texto_para_nlp(resumen, actor_id=actor_id)

    resultado = _classify_text(resumen)
    if not resultado:
        logger.warning(f"[topic_detection] {actor_id}: clasificación fallida")
        return []

    # El router HF puede devolver dos formatos:
    # - Nuevo: [{label, score}, ...]  (lista ordenada por score)
    # - Viejo: {labels: [...], scores: [...]}
    if isinstance(resultado, list):
        labels = [item["label"] for item in resultado]
        scores = [item["score"] for item in resultado]
    else:
        labels = resultado.get("labels", [])
        scores = resultado.get("scores", [])
    total_score = sum(scores) or 1.0

    top_temas = [
        {
            "tema": label,
            "score": round(score, 4),
            "peso_relativo": round(score / total_score * 100, 1),
        }
        for label, score in zip(labels[:top_n], scores[:top_n])
    ]

    guardar_temas_supabase(actor_id, top_temas)

    logger.info(
        f"[topic_detection] {actor_id}: {len(posts)} posts → muestra {min(len(posts), SAMPLE_SIZE)}, "
        f"top tema: '{top_temas[0]['tema']}' ({top_temas[0]['peso_relativo']}%)"
    )
    return top_temas


def get_topic_drift(actor_id: str, dias: int = None) -> dict:
    """
    Detecta si un actor cambió su agenda temática en el período indicado.

    Compara los temas de la primera mitad del período contra la segunda mitad,
    usando los snapshots guardados en actor_topics ordenados por calculado_en.

    Parámetros:
        actor_id: ID del actor en seed_actors.json
        dias:     período a analizar en días (default: TOPIC_DRIFT_DIAS de config)

    Devuelve:
        Dict con topics_antes, topics_ahora, temas_nuevos, temas_abandonados y hay_drift.
        Dict vacío {} si no hay suficientes datos (< 2 snapshots distintos).
    """
    from config import TOPIC_DRIFT_DIAS

    if dias is None:
        dias = TOPIC_DRIFT_DIAS

    if not SUPABASE_URL or not SUPABASE_KEY:
        logger.warning("[topic_detection] get_topic_drift: Supabase no configurado")
        return {}

    try:
        client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY)
        resp = (
            client.table("actor_topics")
            .select("tema, score, calculado_en")
            .eq("actor_id", actor_id)
            .order("calculado_en")
            .execute()
        )
        registros = resp.data or []
    except Exception as e:
        logger.error(f"[topic_detection] get_topic_drift: error leyendo actor_topics para {actor_id}: {e}")
        return {}

    if not registros:
        logger.warning(f"[topic_detection] get_topic_drift: sin datos en actor_topics para {actor_id}")
        return {}

    # Agrupar por timestamp de cálculo para identificar snapshots distintos
    snapshots: dict[str, list[dict]] = {}
    for r in registros:
        ts = r["calculado_en"]
        snapshots.setdefault(ts, []).append({"tema": r["tema"], "score": r["score"]})

    timestamps_ordenados = sorted(snapshots.keys())

    if len(timestamps_ordenados) < 2:
        logger.warning(
            f"[topic_detection] get_topic_drift: {actor_id} tiene solo 1 snapshot, "
            "se necesitan al menos 2 para detectar drift"
        )
        return {}

    # Dividir en dos mitades temporales
    mitad = len(timestamps_ordenados) // 2
    ts_primera_mitad = timestamps_ordenados[:mitad]
    ts_segunda_mitad = timestamps_ordenados[mitad:]

    def _top3(ts_list: list[str]) -> list[dict]:
        """Agrega scores por tema en los timestamps dados y devuelve el top 3."""
        acumulado: dict[str, float] = {}
        for ts in ts_list:
            for item in snapshots[ts]:
                acumulado[item["tema"]] = acumulado.get(item["tema"], 0.0) + item["score"]
        ordenados = sorted(acumulado.items(), key=lambda x: x[1], reverse=True)
        return [{"tema": t, "score": round(s, 4)} for t, s in ordenados[:3]]

    topics_antes = _top3(ts_primera_mitad)
    topics_ahora = _top3(ts_segunda_mitad)

    temas_antes_set = {t["tema"] for t in topics_antes}
    temas_ahora_set = {t["tema"] for t in topics_ahora}

    temas_nuevos = sorted(temas_ahora_set - temas_antes_set)
    temas_abandonados = sorted(temas_antes_set - temas_ahora_set)
    hay_drift = bool(temas_nuevos or temas_abandonados)

    logger.info(
        f"[topic_detection] get_topic_drift: {actor_id} — "
        f"drift={'SÍ' if hay_drift else 'NO'}, "
        f"nuevos={temas_nuevos}, abandonados={temas_abandonados}"
    )

    return {
        "actor_id": actor_id,
        "topics_antes": topics_antes,
        "topics_ahora": topics_ahora,
        "temas_nuevos": temas_nuevos,
        "temas_abandonados": temas_abandonados,
        "hay_drift": hay_drift,
    }


def get_topics_exclusivos(top_n: int = 5) -> list[dict]:
    """
    Devuelve los temas que habla un único actor y nadie más (posicion <= 3).

    Lee actor_topics de todos los actores y filtra los temas que solo aparecen
    en un actor. Cruza con seed_actors.json para obtener el nombre del actor.

    Parámetros:
        top_n: cuántos resultados devolver (default: 5), ordenados por score desc

    Devuelve:
        Lista de dicts con tema, actor_id, actor_nombre y score.
        Lista vacía [] si no hay datos o Supabase no está configurado.
    """
    import json
    from config import SEED_ACTORS_PATH

    if not SUPABASE_URL or not SUPABASE_KEY:
        logger.warning("[topic_detection] get_topics_exclusivos: Supabase no configurado")
        return []

    # Cargar nombres de actores desde seed_actors.json
    try:
        with open(SEED_ACTORS_PATH, encoding="utf-8") as f:
            seed = json.load(f)
        nombres_por_id = {a["id"]: a["nombre"] for a in seed["actors"]}
    except Exception as e:
        logger.error(f"[topic_detection] get_topics_exclusivos: error leyendo seed_actors.json: {e}")
        return []

    # Leer actor_topics con paginación (posicion <= 3)
    try:
        from utils.db import paginar
        registros = paginar(
            tabla="actor_topics",
            columnas="actor_id, tema, score",
            filtros={"posicion": ("lte", 3)},
        )
    except Exception as e:
        logger.error(f"[topic_detection] get_topics_exclusivos: error leyendo actor_topics: {e}")
        return []

    if not registros:
        logger.warning("[topic_detection] get_topics_exclusivos: actor_topics vacío")
        return []

    # Para cada tema, recolectar actores distintos y su score máximo
    temas_actores: dict[str, list[dict]] = {}
    for r in registros:
        tema = r["tema"]
        temas_actores.setdefault(tema, []).append(
            {"actor_id": r["actor_id"], "score": r["score"]}
        )

    # Filtrar temas con un único actor
    exclusivos = []
    for tema, actores in temas_actores.items():
        actor_ids_distintos = {a["actor_id"] for a in actores}
        if len(actor_ids_distintos) == 1:
            actor_id = next(iter(actor_ids_distintos))
            score_max = max(a["score"] for a in actores)
            exclusivos.append({
                "tema": tema,
                "actor_id": actor_id,
                "actor_nombre": nombres_por_id.get(actor_id, actor_id),
                "score": round(score_max, 4),
            })

    exclusivos.sort(key=lambda x: x["score"], reverse=True)
    resultado = exclusivos[:top_n]

    logger.info(
        f"[topic_detection] get_topics_exclusivos: {len(exclusivos)} temas exclusivos encontrados, "
        f"devolviendo top {len(resultado)}"
    )
    return resultado


def get_topics_transversales(min_espectros: int = None) -> list[dict]:
    """
    Devuelve los temas que unen a actores de distintos espectros ideológicos (posicion <= 2).

    Cruza actor_topics con el espectro de cada actor (ideology_labels.json) y filtra
    los temas que aparecen en al menos min_espectros espectros distintos.

    Parámetros:
        min_espectros: mínimo de espectros distintos para considerar un tema transversal
                       (default: TOPIC_MIN_ACTORES_TRANSVERSAL de config)

    Devuelve:
        Lista de dicts con tema, espectros, n_espectros y actores, ordenada por n_espectros desc.
        Lista vacía [] si no hay datos o Supabase no está configurado.
    """
    import json
    from config import SEED_ACTORS_PATH, IDEOLOGY_LABELS_PATH, TOPIC_MIN_ACTORES_TRANSVERSAL

    if min_espectros is None:
        min_espectros = TOPIC_MIN_ACTORES_TRANSVERSAL

    if not SUPABASE_URL or not SUPABASE_KEY:
        logger.warning("[topic_detection] get_topics_transversales: Supabase no configurado")
        return []

    # Cargar nombres de actores desde seed_actors.json
    try:
        with open(SEED_ACTORS_PATH, encoding="utf-8") as f:
            seed = json.load(f)
        nombres_por_id = {a["id"]: a["nombre"] for a in seed["actors"]}
    except Exception as e:
        logger.error(f"[topic_detection] get_topics_transversales: error leyendo seed_actors.json: {e}")
        return []

    # Cargar espectros desde ideology_labels.json
    try:
        with open(IDEOLOGY_LABELS_PATH, encoding="utf-8") as f:
            ideology = json.load(f)
        # Estructura: {"labels": {actor_id: {"espectro": "ED", ...}, ...}}
        labels = ideology.get("labels", ideology)
        espectro_por_id = {
            actor_id: datos.get("espectro", "")
            for actor_id, datos in labels.items()
        }
    except Exception as e:
        logger.error(f"[topic_detection] get_topics_transversales: error leyendo ideology_labels.json: {e}")
        return []

    # Leer actor_topics con paginación (posicion <= 2)
    try:
        from utils.db import paginar
        registros = paginar(
            tabla="actor_topics",
            columnas="actor_id, tema, score",
            filtros={"posicion": ("lte", 2)},
        )
    except Exception as e:
        logger.error(f"[topic_detection] get_topics_transversales: error leyendo actor_topics: {e}")
        return []

    if not registros:
        logger.warning("[topic_detection] get_topics_transversales: actor_topics vacío")
        return []

    # Para cada tema, recolectar actores con su espectro
    temas_info: dict[str, dict] = {}
    for r in registros:
        actor_id = r["actor_id"]
        tema = r["tema"]
        espectro = espectro_por_id.get(actor_id, "")
        if not espectro:
            continue  # actor sin espectro definido, se omite

        if tema not in temas_info:
            temas_info[tema] = {"espectros": set(), "actores": []}

        temas_info[tema]["espectros"].add(espectro)
        # Evitar duplicar el mismo actor por posicion 1 y 2
        actor_ids_existentes = {a["id"] for a in temas_info[tema]["actores"]}
        if actor_id not in actor_ids_existentes:
            temas_info[tema]["actores"].append({
                "id": actor_id,
                "nombre": nombres_por_id.get(actor_id, actor_id),
                "espectro": espectro,
            })

    # Filtrar por mínimo de espectros distintos
    transversales = []
    for tema, info in temas_info.items():
        n_espectros = len(info["espectros"])
        if n_espectros >= min_espectros:
            actores_ordenados = sorted(info["actores"], key=lambda a: a["espectro"])
            transversales.append({
                "tema": tema,
                "espectros": sorted(info["espectros"]),
                "n_espectros": n_espectros,
                "actores": actores_ordenados,
            })

    transversales.sort(key=lambda x: x["n_espectros"], reverse=True)

    logger.info(
        f"[topic_detection] get_topics_transversales: {len(transversales)} temas "
        f"con >= {min_espectros} espectros distintos"
    )
    return transversales