""" synchrony_detector.py — Detección de sincronización temática entre actores. Basado en: "Unveiling political influence through social media: network and causal dynamics in the 2022 French presidential election" (Frontiers in Political Science, 2025). Algoritmo: Para cada par de actores (A, B): 1. Tomar sus posts de los últimos 90 días 2. Dividir el tiempo en ventanas de N horas 3. Para cada ventana: calcular similitud semántica entre posts de A y B (coseno entre embeddings promedio de cada grupo) 4. Contar ventanas donde similitud > umbral 5. Calcular línea base: ventanas esperadas por azar (proporción de ventanas con posts * total) 6. Ratio real/esperado > 2.0 → sincronización significativa Advertencia: sincronización temática NO implica coordinación intencional. Es una señal para investigar, no una conclusión. """ import logging import math import re from collections import defaultdict from datetime import datetime, timedelta, timezone from itertools import combinations import networkx as nx import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from utils.db import paginar as _paginar from config import ( SEED_ACTORS_PATH, IDEOLOGY_LABELS_PATH, SUPABASE_URL, SUPABASE_KEY, SUPABASE_SERVICE_KEY, LOG_FORMAT, LOG_DATE_FORMAT, LOG_LEVEL, ) logging.basicConfig(format=LOG_FORMAT, datefmt=LOG_DATE_FORMAT) logger = logging.getLogger("synchrony_detector") logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO)) # ── Helpers ─────────────────────────────────────────────────────────────────── def _coseno(v1: list[float], v2: list[float]) -> float: """Similitud coseno entre dos vectores. Devuelve 0.0 si alguno es nulo.""" if not v1 or not v2: return 0.0 dot = sum(a * b for a, b in zip(v1, v2)) norm1 = math.sqrt(sum(a * a for a in v1)) norm2 = math.sqrt(sum(b * b for b in v2)) if norm1 == 0.0 or norm2 == 0.0: return 0.0 return dot / (norm1 * norm2) def _promedio_vectores(vectores: list[list[float]]) -> list[float]: """Calcula el vector promedio de una lista. Devuelve [] si la lista está vacía.""" if not vectores: return [] n = len(vectores) dim = len(vectores[0]) return [sum(v[i] for v in vectores) / n for i in range(dim)] def _parsear_fecha(fecha_str: str | None) -> datetime | None: """Parsea una fecha ISO 8601 a datetime UTC. Devuelve None si falla.""" if not fecha_str: return None try: dt = datetime.fromisoformat(fecha_str.replace("Z", "+00:00")) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt except (ValueError, AttributeError): return None def _leer_posts_con_embeddings(dias: int = 90) -> list[dict]: """ Lee posts de Supabase con embeddings, filtrando los últimos N días. Estrategia de caché: 1. Fetch todos los posts del período (paginado). 2. Consulta post_embeddings para los IDs obtenidos (en chunks de 500). 3. Genera embeddings solo para posts sin caché via HF API. 4. Persiste los nuevos embeddings en post_embeddings (batches de 100). 5. Devuelve todos los posts con embedding adjunto. """ if not SUPABASE_URL or not SUPABASE_KEY: return [] try: from supabase import create_client client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) corte = (datetime.now(timezone.utc) - timedelta(days=dias)).isoformat() # 1. Fetch todos los posts del período (paginado) posts = _paginar( lambda: client.table("posts") .select("id, actor_id, titulo, resumen, publicado") .gte("publicado", corte) ) except Exception as e: logger.error(f"[synchrony] error leyendo posts: {e}") return [] if not posts: logger.info("[synchrony] sin posts en el período analizado") return [] # 2. Leer embeddings cacheados de post_embeddings (en chunks de 500) cached: dict[str, list[float]] = {} post_ids = [p["id"] for p in posts] try: CHUNK = 500 for i in range(0, len(post_ids), CHUNK): chunk = post_ids[i:i + CHUNK] resp = client.table("post_embeddings").select("post_id, embedding").in_("post_id", chunk).execute() for row in (resp.data or []): cached[row["post_id"]] = row["embedding"] logger.info(f"[synchrony] {len(cached)}/{len(posts)} embeddings en caché") except Exception as e: # La tabla puede no existir en instancias viejas — fallback a generar todo logger.warning(f"[synchrony] no se pudo leer post_embeddings: {e} — generando todos") # 3. Generar embeddings solo para posts sin caché sin_cache = [p for p in posts if p["id"] not in cached] if sin_cache: logger.info(f"[synchrony] generando embeddings para {len(sin_cache)} posts nuevos...") try: from processing.embeddings import get_embeddings textos = [ f"{p.get('titulo', '')} {p.get('resumen', '')}".strip() for p in sin_cache ] nuevos = get_embeddings(textos) for post, emb in zip(sin_cache, nuevos): if emb: cached[post["id"]] = emb except Exception as e: logger.error(f"[synchrony] error generando embeddings: {e}") # 4. Persistir nuevos embeddings en batches de 100 ahora = datetime.now(timezone.utc).isoformat() filas = [ {"post_id": p["id"], "embedding": cached[p["id"]], "calculado_en": ahora} for p in sin_cache if p["id"] in cached ] if filas: try: BATCH = 100 for i in range(0, len(filas), BATCH): client.table("post_embeddings").upsert( filas[i:i + BATCH], on_conflict="post_id" ).execute() logger.info(f"[synchrony] {len(filas)} embeddings guardados en post_embeddings") except Exception as e: logger.warning(f"[synchrony] no se pudo guardar embeddings: {e}") # 5. Adjuntar embeddings a cada post y devolver solo los que tienen embedding result = [] for p in posts: emb = cached.get(p["id"]) if emb: p["embedding"] = emb result.append(p) logger.info(f"[synchrony] {len(result)}/{len(posts)} posts con embedding disponible") return result def _agrupar_por_ventana( posts: list[dict], window_hours: int, ) -> dict[tuple[str, int], list[list[float]]]: """ Agrupa embeddings por (actor_id, numero_de_ventana). Devuelve dict: {(actor_id, ventana): [embedding1, embedding2, ...]} """ grupos: dict[tuple[str, int], list[list[float]]] = defaultdict(list) for post in posts: fecha = _parsear_fecha(post.get("publicado")) if not fecha: continue # Número de ventana: horas desde epoch / window_hours ventana = int(fecha.timestamp() / 3600 / window_hours) grupos[(post["actor_id"], ventana)].append(post["embedding"]) return grupos def _cargar_metadatos_actores() -> dict[str, dict]: """ Carga categoria y espectro de cada actor desde archivos JSON locales. Devuelve dict {actor_id: {"categoria": str, "espectro": str}}. """ try: import json with open(SEED_ACTORS_PATH, encoding="utf-8") as f: actores = json.load(f)["actors"] cat_map = {a["id"]: a.get("categoria", "") for a in actores} with open(IDEOLOGY_LABELS_PATH, encoding="utf-8") as f: labels = json.load(f)["labels"] return { actor_id: { "categoria": cat_map.get(actor_id, ""), "espectro": info.get("espectro", "C"), } for actor_id, info in labels.items() } except Exception as e: logger.warning(f"[synchrony] no se pudo cargar metadatos: {e}") return {} # ── Funciones públicas ──────────────────────────────────────────────────────── def detect_synchrony( min_posts_per_window: int = 2, similarity_threshold: float = 0.65, time_window_hours: int = 6, dias: int = 90, ) -> list[dict]: """ Detecta pares de actores con sincronización temática significativa. Para cada par (A, B) calcula: - ventanas_sincronizadas: ventanas donde ambos publicaron y similitud > umbral - ventanas_esperadas: línea base estadística (azar) - ratio: real / esperado — ratio > 2.0 es significativo Parámetros: min_posts_per_window: mínimo de posts para que una ventana cuente similarity_threshold: similitud coseno mínima para considerar sincronización time_window_hours: tamaño de la ventana temporal en horas dias: cuántos días hacia atrás analizar Devuelve: Lista de dicts ordenada por ratio descendente. Lista vacía si no hay datos. ADVERTENCIA: sincronización != coordinación. Es una señal, no una conclusión. """ posts = _leer_posts_con_embeddings(dias=dias) if not posts: logger.info("[synchrony] sin posts con embeddings para analizar") return [] meta = _cargar_metadatos_actores() grupos = _agrupar_por_ventana(posts, time_window_hours) # Actores con posts en el período actores_con_posts = list({actor_id for actor_id, _ in grupos.keys()}) if len(actores_con_posts) < 2: logger.info("[synchrony] menos de 2 actores con posts — sin pares para analizar") return [] # Ventanas en que cada actor tiene posts suficientes ventanas_por_actor: dict[str, set[int]] = defaultdict(set) embeddings_por_ventana: dict[tuple[str, int], list[float]] = {} for (actor_id, ventana), embeddings in grupos.items(): if len(embeddings) >= min_posts_per_window: ventanas_por_actor[actor_id].add(ventana) embeddings_por_ventana[(actor_id, ventana)] = _promedio_vectores(embeddings) total_ventanas = (dias * 24) // time_window_hours MIN_POSTS = 30 # FILTRO 4: mínimo de posts por actor en el período # Contar posts totales por actor en el período analizado posts_por_actor: dict[str, int] = defaultdict(int) for post in posts: posts_por_actor[post["actor_id"]] += 1 resultados = [] pares = list(combinations(actores_con_posts, 2)) logger.info(f"[synchrony] analizando {len(pares)} pares de actores...") for actor_a, actor_b in pares: meta_a = meta.get(actor_a, {}) meta_b = meta.get(actor_b, {}) # FILTRO 1: Excluir pares donde ambos son medios (ruido estructural) if meta_a.get("categoria") == "medio" and meta_b.get("categoria") == "medio": continue # FILTRO 4: Mínimo de posts para que el par sea analizable if posts_por_actor[actor_a] < MIN_POSTS or posts_por_actor[actor_b] < MIN_POSTS: continue ventanas_a = ventanas_por_actor[actor_a] ventanas_b = ventanas_por_actor[actor_b] ventanas_comunes = ventanas_a & ventanas_b if not ventanas_comunes: continue # Similitud en ventanas comunes sincronizadas = 0 for v in ventanas_comunes: emb_a = embeddings_por_ventana.get((actor_a, v), []) emb_b = embeddings_por_ventana.get((actor_b, v), []) if emb_a and emb_b and _coseno(emb_a, emb_b) >= similarity_threshold: sincronizadas += 1 # Línea base real: prob(A publica) * prob(B publica) * total_ventanas p_a = len(ventanas_a) / max(total_ventanas, 1) p_b = len(ventanas_b) / max(total_ventanas, 1) esperadas = p_a * p_b * total_ventanas if esperadas == 0: continue ratio = sincronizadas / esperadas # FILTRO 2: Tasa base alta → ambos publican muy frecuentemente, # la coincidencia de ventanas es esperable y no indica sincronización tasa_comunes = len(ventanas_comunes) / max(total_ventanas, 1) tasa_base_alta = tasa_comunes > 0.3 # FILTRO 3: Mismo espectro → umbral más estricto # (medios del mismo bloque suelen cubrir los mismos temas por definición) # En Argentina toda la agenda política gira sobre los mismos 3-4 temas → # umbrales más altos que el paper francés para reducir falsos positivos. mismo_espectro = meta_a.get("espectro") == meta_b.get("espectro") ratio_min = 3.5 if mismo_espectro else 2.5 if tasa_base_alta: ratio_min = max(ratio_min, 4.0) es_significativo = ratio >= ratio_min and sincronizadas >= 3 # es_ruido_probable: señal de que la sincronización puede ser espuria es_ruido_probable = (mismo_espectro and ratio < 3.5) or (tasa_base_alta and ratio < 4.0) resultados.append({ "actor_a": actor_a, "actor_b": actor_b, "sync_score": round(_coseno( _promedio_vectores([embeddings_por_ventana.get((actor_a, v), []) for v in ventanas_comunes if embeddings_por_ventana.get((actor_a, v))]), _promedio_vectores([embeddings_por_ventana.get((actor_b, v), []) for v in ventanas_comunes if embeddings_por_ventana.get((actor_b, v))]), ), 4), "ventanas_sincronizadas": sincronizadas, "ventanas_esperadas": round(esperadas, 2), "ratio": round(ratio, 3), "ratio_min_usado": ratio_min, "es_significativo": es_significativo, "es_ruido_probable": es_ruido_probable, }) resultados.sort(key=lambda x: x["ratio"], reverse=True) significativos = sum(1 for r in resultados if r["es_significativo"]) logger.info( f"[synchrony] {len(resultados)} pares analizados, " f"{significativos} con sincronización significativa (umbral dinámico 2.5–4.0)" ) return resultados def get_synchrony_network( min_posts_per_window: int = 2, similarity_threshold: float = 0.65, time_window_hours: int = 6, ) -> nx.Graph: """ Construye un grafo NO dirigido donde los nodos son actores y las aristas representan pares con sincronización significativa (ratio ≥ 2.0). El grosor de la arista (weight) es proporcional al ratio de sincronización. Umbral dinámico: 2.5× para espectros distintos, 3.5× para mismo espectro. Devuelve: nx.Graph — puede estar vacío si no hay pares significativos. """ resultados = detect_synchrony( min_posts_per_window=min_posts_per_window, similarity_threshold=similarity_threshold, time_window_hours=time_window_hours, ) G = nx.Graph() for r in resultados: if r["es_significativo"]: G.add_edge( r["actor_a"], r["actor_b"], weight=r["ratio"], sync_score=r["sync_score"], ventanas=r["ventanas_sincronizadas"], ) logger.info(f"[synchrony] red de sincronización: {G.number_of_nodes()} nodos, {G.number_of_edges()} aristas") return G def guardar_en_supabase(resultados: list[dict]) -> None: """ Guarda los resultados en la tabla synchrony_scores. Solo guarda pares significativos para mantener la tabla manejable. """ if not resultados or not SUPABASE_URL or not SUPABASE_KEY: return significativos = [r for r in resultados if r["es_significativo"]] if not significativos: logger.info("[synchrony] sin pares significativos para guardar") return try: from supabase import create_client client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) ahora = datetime.now(timezone.utc).isoformat() rows = [ { "actor_a": r["actor_a"], "actor_b": r["actor_b"], "sync_score": r["sync_score"], "ratio": r["ratio"], "ventanas_sincronizadas": r["ventanas_sincronizadas"], "es_ruido_probable": r.get("es_ruido_probable", False), "timestamp_calculado": ahora, } for r in significativos ] try: client.table("synchrony_scores").upsert( rows, on_conflict="actor_a,actor_b" ).execute() logger.info(f"[synchrony] {len(rows)} pares guardados en Supabase") except Exception as e_inner: # Si falla por columna es_ruido_probable inexistente, reintentar sin ella if "es_ruido_probable" in str(e_inner).lower() or "column" in str(e_inner).lower(): logger.warning( "[synchrony] columna es_ruido_probable no encontrada — " "guardando sin ese campo. Ejecutá el DDL en ESTADO.md para activarla." ) rows_sin_ruido = [ {k: v for k, v in row.items() if k != "es_ruido_probable"} for row in rows ] client.table("synchrony_scores").upsert( rows_sin_ruido, on_conflict="actor_a,actor_b" ).execute() logger.info(f"[synchrony] {len(rows_sin_ruido)} pares guardados (sin es_ruido_probable)") else: raise except Exception as e: logger.warning(f"[synchrony] error guardando en Supabase: {e}")