Spaces:
Running
Running
| """ | |
| 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 | |