""" 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