""" emerging_actors.py — Detección de actores emergentes no mapeados en seed_actors.json. Lee todos los posts de Supabase, extrae @handles mencionados y filtra los que NO están en seed_actors.json. Útil para descubrir figuras nuevas que están ganando relevancia en el ecosistema político argentino. El usuario decide manualmente si incorporarlos — nunca se agregan automáticamente. Métricas calculadas por handle emergente: - menciones: total histórico - menciones_recientes: menciones en los últimos EMERGING_VELOCIDAD_DIAS días - velocidad: menciones_recientes / EMERGING_VELOCIDAD_DIAS (menciones/día) - espectros_que_mencionan: espectros únicos de los actores que lo citan - es_consenso: True si lo mencionan actores de >= 3 espectros distintos """ import json import logging import re from collections import defaultdict from datetime import datetime, timedelta, timezone from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).parent.parent)) from config import ( SEED_ACTORS_PATH, IDEOLOGY_LABELS_PATH, SUPABASE_URL, SUPABASE_KEY, SUPABASE_SERVICE_KEY, EMERGING_MIN_MENTIONS, EMERGING_TOP_N, EMERGING_VELOCIDAD_DIAS, LOG_FORMAT, LOG_DATE_FORMAT, LOG_LEVEL, ) from utils.db import paginar as _paginar logging.basicConfig(format=LOG_FORMAT, datefmt=LOG_DATE_FORMAT) logger = logging.getLogger("emerging_actors") logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO)) # Espectros válidos en orden canónico _ORDEN_ESPECTROS = ["ED", "CD", "C", "CI", "EI"] def _cargar_handles_conocidos() -> set[str]: """Devuelve el conjunto de handles Twitter conocidos (sin @, lowercase).""" with open(SEED_ACTORS_PATH, encoding="utf-8") as f: actores = json.load(f)["actors"] return { a["twitter"].lstrip("@").lower() for a in actores if a.get("twitter") } def _cargar_espectros() -> dict[str, str]: """ Devuelve un dict {actor_id: espectro} leyendo ideology_labels.json. Si el archivo no existe o falla, devuelve dict vacío y loguea el error. """ try: with open(IDEOLOGY_LABELS_PATH, encoding="utf-8") as f: datos = json.load(f) return { actor_id: info["espectro"] for actor_id, info in datos.get("labels", {}).items() if "espectro" in info } except FileNotFoundError: logger.error(f"[emerging_actors] ideology_labels.json no encontrado en {IDEOLOGY_LABELS_PATH}") return {} except Exception as e: logger.error(f"[emerging_actors] error leyendo ideology_labels.json: {e}") return {} def _parsear_publicado(valor: str | None) -> datetime | None: """ Parsea un string ISO 8601 con timezone a datetime aware. Devuelve None si el valor es vacío o no parseable. """ if not valor: return None try: # Python 3.11+ acepta fromisoformat con 'Z'; para 3.10 reemplazamos ts = valor.replace("Z", "+00:00") return datetime.fromisoformat(ts) except Exception as e: logger.warning(f"[emerging_actors] no se pudo parsear fecha '{valor}': {e}") return None def get_emerging_actors(min_mentions: int = EMERGING_MIN_MENTIONS, top_n: int = EMERGING_TOP_N) -> list[dict]: """ Detecta handles mencionados en posts que no están en seed_actors.json. Lee todos los posts de Supabase, extrae @mentions de títulos y resúmenes, descarta los handles conocidos y devuelve un ranking por velocidad reciente. Parámetros: min_mentions: mínimo de menciones totales para aparecer en el ranking top_n: máximo de actores a devolver, ordenados por menciones_recientes desc Devuelve: Lista de dicts ordenada por menciones_recientes desc: [ { "handle": str, # con @ "menciones": int, # total histórico "menciones_recientes": int, # en últimos EMERGING_VELOCIDAD_DIAS días "velocidad": float, # menciones_recientes / EMERGING_VELOCIDAD_DIAS "mencionado_por": list[str], # actor_ids "espectros_que_mencionan": list[str], # espectros únicos, orden canónico "n_espectros": int, "es_consenso": bool, # True si >= 3 espectros distintos } ] Lista vacía si no hay datos o Supabase no está configurado. """ if not SUPABASE_URL or not SUPABASE_KEY: logger.warning("[emerging_actors] Supabase no configurado") return [] try: from supabase import create_client client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) posts = _paginar( lambda: client.table("posts").select("actor_id, titulo, resumen, publicado") ) except Exception as e: logger.error(f"[emerging_actors] error leyendo Supabase: {e}") return [] if not posts: logger.info("[emerging_actors] sin posts en Supabase") return [] handles_conocidos = _cargar_handles_conocidos() espectros_por_actor = _cargar_espectros() # Umbral de tiempo para menciones recientes ahora = datetime.now(timezone.utc) limite_reciente = ahora - timedelta(days=EMERGING_VELOCIDAD_DIAS) # Acumuladores por handle conteo_total: dict[str, int] = defaultdict(int) conteo_reciente: dict[str, int] = defaultdict(int) mencionado_por: dict[str, set] = defaultdict(set) # actor_ids espectros_por_handle: dict[str, set] = defaultdict(set) # espectros de quienes mencionan for post in posts: texto = f"{post.get('titulo', '')} {post.get('resumen', '')}" handles_en_post = re.findall(r"@(\w+)", texto.lower()) if not handles_en_post: continue actor_id = post.get("actor_id", "desconocido") espectro_autor = espectros_por_actor.get(actor_id) fecha_post = _parsear_publicado(post.get("publicado")) es_reciente = fecha_post is not None and fecha_post >= limite_reciente for handle in handles_en_post: if handle in handles_conocidos: continue # ya está mapeado conteo_total[handle] += 1 mencionado_por[handle].add(actor_id) if es_reciente: conteo_reciente[handle] += 1 if espectro_autor: espectros_por_handle[handle].add(espectro_autor) # Filtrar por mínimo de menciones totales y construir resultado emergentes = [] for handle, total in conteo_total.items(): if total < min_mentions: continue recientes = conteo_reciente[handle] velocidad = round(recientes / EMERGING_VELOCIDAD_DIAS, 2) # Espectros en orden canónico espectros_set = espectros_por_handle[handle] espectros_ordenados = [e for e in _ORDEN_ESPECTROS if e in espectros_set] n_espectros = len(espectros_ordenados) emergentes.append({ "handle": f"@{handle}", "menciones": total, "menciones_recientes": recientes, "velocidad": velocidad, "mencionado_por": sorted(mencionado_por[handle]), "espectros_que_mencionan": espectros_ordenados, "n_espectros": n_espectros, "es_consenso": n_espectros >= 3, }) # Ordenar por menciones_recientes desc, desempate por menciones totales desc emergentes.sort(key=lambda x: (x["menciones_recientes"], x["menciones"]), reverse=True) emergentes = emergentes[:top_n] consensos = sum(1 for e in emergentes if e["es_consenso"]) logger.info( f"[emerging_actors] {len(emergentes)} actores emergentes " f"con >= {min_mentions} menciones totales (de {len(posts)} posts analizados, top {top_n}). " f"Ventana reciente: {EMERGING_VELOCIDAD_DIAS}d. Consensos: {consensos}." ) return emergentes