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