Blame for processing/amplification_detector.py

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