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