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