""" amplification_detector.py — Detección de cadenas de amplificación temporal. Una cadena de amplificación es: actor A publica algo → actor B lo retweetea en menos de AMPLIFICATION_WINDOW_MINUTES minutos → actor C lo replica en la misma ventana → etc. Esto indica coordinación o prensa orquestada — algo que no se puede detectar scrolleando Twitter pero sí con un grafo dirigido temporal. Función pública principal: detectar_cadenas(dias) -> list[dict] """ import json import logging 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, SUPABASE_URL, SUPABASE_KEY, SUPABASE_SERVICE_KEY, AMPLIFICATION_WINDOW_MINUTES, AMPLIFICATION_MIN_CHAIN, 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("amplification_detector") logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO)) # ── Helpers ──────────────────────────────────────────────────────────────────── def _cargar_actores_conocidos() -> dict[str, str]: """ Devuelve dict {actor_id: nombre} desde seed_actors.json. También incluye un mapa de handle → actor_id para cruzar con interaccion_con. """ try: with open(SEED_ACTORS_PATH, encoding="utf-8") as f: actores = json.load(f)["actors"] return {a["id"]: a["nombre"] for a in actores} except Exception as e: logger.error(f"[amplification] error cargando seed_actors.json: {e}") return {} def _parsear_ts(valor: str | None) -> datetime | None: """Parsea un string ISO 8601 con timezone a datetime aware. None si falla.""" if not valor: return None try: return datetime.fromisoformat(valor.replace("Z", "+00:00")) except (ValueError, AttributeError): return None def _leer_posts_con_interacciones(dias: int) -> list[dict]: """ Lee posts con interaccion_con no nulo de los últimos `dias` días. Devuelve lista con actor_id, publicado, interaccion_con, tipo_interaccion. """ if not SUPABASE_URL or not SUPABASE_KEY: logger.warning("[amplification] Supabase no configurado") return [] try: from supabase import create_client client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) from_date = (datetime.now(timezone.utc) - timedelta(days=dias)).isoformat() return _paginar( lambda: ( client.table("posts") .select("actor_id, publicado, interaccion_con, tipo_interaccion") .gte("publicado", from_date) .not_.is_("interaccion_con", "null") ) ) except Exception as e: logger.error(f"[amplification] error leyendo Supabase: {e}") return [] # ── Algoritmo de cadenas ─────────────────────────────────────────────────────── def _construir_eventos(posts: list[dict], actores_validos: set[str]) -> list[dict]: """ Convierte posts crudos en eventos ordenados por timestamp. Filtra los que tengan receptor fuera del seed o timestamps inválidos. """ eventos = [] for post in posts: origen = post.get("actor_id") destino = post.get("interaccion_con") tipo = (post.get("tipo_interaccion") or "").lower() ts = _parsear_ts(post.get("publicado")) if not origen or not destino or not ts: continue if origen not in actores_validos or destino not in actores_validos: continue if tipo not in ("retweet", "reply"): continue eventos.append({ "origen": origen, "destino": destino, "tipo": tipo, "ts": ts, }) eventos.sort(key=lambda e: e["ts"]) return eventos def _detectar_eslabones(eventos: list[dict]) -> list[tuple]: """ Detecta eslabones A→B donde B amplifica algo de A dentro de la ventana temporal. Un eslabón existe cuando: - B retweetea o replica a A - El post de A existe en la ventana de AMPLIFICATION_WINDOW_MINUTES antes del post de B Devuelve lista de (origen, destino, ts_origen, ts_destino, tipo). """ ventana = timedelta(minutes=AMPLIFICATION_WINDOW_MINUTES) # Índice de posts por actor: {actor_id: [timestamps de sus posts]} posts_por_actor: dict[str, list[datetime]] = defaultdict(list) for ev in eventos: # El "destino" es quien fue amplificado: registrar su actividad posts_por_actor[ev["destino"]].append(ev["ts"]) eslabones = [] for ev in eventos: amplificador = ev["origen"] # quien amplifica amplificado = ev["destino"] # a quien amplificaron ts_amplificacion = ev["ts"] # Buscar si el actor amplificado publicó algo dentro de la ventana previa ts_posibles = [ ts for ts in posts_por_actor.get(amplificado, []) if ts_amplificacion - ventana <= ts < ts_amplificacion ] if ts_posibles: ts_original = min(ts_posibles) # el más temprano en la ventana eslabones.append(( amplificado, # quien publicó originalmente amplificador, # quien amplificó ts_original, ts_amplificacion, ev["tipo"], )) return eslabones def _unir_eslabones_en_cadenas(eslabones: list[tuple]) -> list[list[tuple]]: """ Une eslabones contiguos para formar cadenas. Si A→B y B→C existen con timestamps compatibles, forma la cadena A→B→C. Devuelve lista de cadenas, donde cada cadena es lista de eslabones (origen, destino, ts_origen, ts_destino, tipo). """ if not eslabones: return [] ventana = timedelta(minutes=AMPLIFICATION_WINDOW_MINUTES) # Para buscar extensiones: {destino: [eslabones que empiezan con ese destino como origen]} por_origen: dict[str, list[tuple]] = defaultdict(list) for esl in eslabones: por_origen[esl[0]].append(esl) cadenas_completas = [] visitados: set[tuple] = set() def _extender(cadena_actual: list[tuple]) -> None: """Recursivamente extiende la cadena hacia adelante.""" ultimo = cadena_actual[-1] ultimo_destino = ultimo[1] ultimo_ts_destino = ultimo[3] continuaciones = [ esl for esl in por_origen.get(ultimo_destino, []) if ultimo_ts_destino <= esl[2] <= ultimo_ts_destino + ventana and esl not in visitados ] if not continuaciones: if len(cadena_actual) >= AMPLIFICATION_MIN_CHAIN: cadenas_completas.append(list(cadena_actual)) return for esl in continuaciones: visitados.add(esl) cadena_actual.append(esl) _extender(cadena_actual) cadena_actual.pop() visitados.discard(esl) for esl in eslabones: if esl not in visitados: visitados.add(esl) _extender([esl]) return cadenas_completas def _cadena_a_dict(cadena: list[tuple], nombres: dict[str, str]) -> dict: """Convierte una cadena de eslabones al formato de retorno público.""" actores_en_cadena = [cadena[0][0]] # primer origen timestamps = [cadena[0][2].isoformat()] tipos = [] for esl in cadena: actores_en_cadena.append(esl[1]) timestamps.append(esl[3].isoformat()) tipos.append(esl[4]) # Deduplicar actores preservando orden (puede haber ciclos) vistos: set[str] = set() actores_unicos = [] ts_unicos = [] for actor, ts in zip(actores_en_cadena, timestamps): if actor not in vistos: vistos.add(actor) actores_unicos.append(actor) ts_unicos.append(ts) ts_inicio = _parsear_ts(ts_unicos[0]) ts_fin = _parsear_ts(ts_unicos[-1]) duracion = int((ts_fin - ts_inicio).total_seconds() / 60) if ts_inicio and ts_fin else 0 tipo_predominante = ( "retweet" if tipos.count("retweet") >= tipos.count("reply") else "reply" ) if tipos else "mixta" return { "cadena": actores_unicos, "timestamps": ts_unicos, "tipo": tipo_predominante, "duracion_minutos": duracion, "nombres": [nombres.get(aid, aid) for aid in actores_unicos], } # ── Funciones públicas ───────────────────────────────────────────────────────── def detectar_cadenas(dias: int = 7) -> list[dict]: """ Detecta cadenas de amplificación temporal en los últimos `dias` días. Una cadena es una secuencia de actores donde cada uno amplifica al anterior (RT o reply) dentro de una ventana de AMPLIFICATION_WINDOW_MINUTES minutos. Parámetros: dias: período de análisis hacia atrás desde ahora Devuelve: Lista de cadenas ordenada por longitud desc. Cada cadena: { "cadena": list[str], # actor_ids en orden temporal "timestamps": list[str], # ISO 8601 por actor "tipo": str, # "retweet" | "reply" | "mixta" "duracion_minutos": int, # desde primer al último evento "nombres": list[str], # nombres legibles } Lista vacía si no hay datos suficientes o Supabase no está configurado. """ nombres = _cargar_actores_conocidos() if not nombres: return [] actores_validos = set(nombres.keys()) posts = _leer_posts_con_interacciones(dias) if not posts: logger.info(f"[amplification] sin posts con interacciones en los últimos {dias} días") return [] eventos = _construir_eventos(posts, actores_validos) logger.info(f"[amplification] {len(eventos)} eventos válidos de {len(posts)} posts") if not eventos: return [] eslabones = _detectar_eslabones(eventos) logger.info(f"[amplification] {len(eslabones)} eslabones detectados") if not eslabones: return [] cadenas = _unir_eslabones_en_cadenas(eslabones) # Ordenar por longitud desc, luego por duracion asc (cadenas largas y rápidas primero) cadenas.sort(key=lambda c: (-len(c), c[-1][3] - c[0][2])) resultado = [_cadena_a_dict(c, nombres) for c in cadenas] logger.info( f"[amplification] {len(resultado)} cadenas detectadas " f"(ventana: {AMPLIFICATION_WINDOW_MINUTES}min, mín: {AMPLIFICATION_MIN_CHAIN} eslabones)" ) return resultado def get_cadenas_summary(dias: int = 7) -> dict: """ Devuelve un resumen de cadenas de amplificación para el dashboard. Parámetros: dias: período de análisis Devuelve: { "total_cadenas": int, "cadena_mas_larga": int, # número de actores "actor_mas_iniciador": dict, # {id, nombre, veces} "actor_mas_amplificador": dict, # {id, nombre, veces} "cadenas": list, # top 5 cadenas más largas } Dict vacío si no hay cadenas. """ cadenas = detectar_cadenas(dias=dias) if not cadenas: return {} # Contar roles iniciadores: dict[str, int] = defaultdict(int) amplificadores: dict[str, int] = defaultdict(int) for cadena in cadenas: actores = cadena["cadena"] if actores: iniciadores[actores[0]] += 1 for actor in actores[1:]: amplificadores[actor] += 1 nombres_map = _cargar_actores_conocidos() def _top(conteo: dict[str, int]) -> dict: if not conteo: return {} aid = max(conteo, key=lambda k: conteo[k]) return {"id": aid, "nombre": nombres_map.get(aid, aid), "veces": conteo[aid]} return { "total_cadenas": len(cadenas), "cadena_mas_larga": max(len(c["cadena"]) for c in cadenas), "actor_mas_iniciador": _top(iniciadores), "actor_mas_amplificador": _top(amplificadores), "cadenas": cadenas[:5], }