Spaces:
Running
Running
| """ | |
| influence_scorer.py — Cálculo de influencia mediante PageRank sobre grafo de menciones. | |
| Construye un grafo dirigido donde: | |
| - Nodos: actores (65 figuras públicas) | |
| - Aristas: menciones de un actor en posts de otro actor | |
| - Peso: cantidad de veces que actor A menciona a actor B | |
| Calcula PageRank con NetworkX y guarda resultados en Supabase (tabla influence_scores). | |
| No usa NLP — solo análisis de grafos sobre datos ya recolectados. | |
| """ | |
| import json | |
| import logging | |
| import math | |
| import re | |
| from datetime import datetime, timedelta, timezone | |
| 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, | |
| GRAPH_PAGERANK_ALPHA, | |
| GRAPH_MIN_EDGE_WEIGHT, | |
| 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("influence_scorer") | |
| logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO)) | |
| # Flag para crear la tabla influence_scores solo una vez por proceso | |
| _TABLA_INFLUENCE_VERIFICADA: bool = False | |
| # ── Helpers ─────────────────────────────────────────────────────────────────── | |
| def _cargar_actores() -> list[dict]: | |
| """Carga y devuelve la lista completa de actores desde seed_actors.json.""" | |
| with open(SEED_ACTORS_PATH, encoding="utf-8") as f: | |
| return json.load(f)["actors"] | |
| def _cargar_etiquetas() -> dict: | |
| """Carga ideology_labels.json y devuelve el dict de etiquetas por actor_id.""" | |
| with open(IDEOLOGY_LABELS_PATH, encoding="utf-8") as f: | |
| return json.load(f)["labels"] | |
| def _handles_a_ids(actores: list[dict]) -> dict[str, str]: | |
| """ | |
| Construye un mapa de handle Twitter (sin @, lowercase) → actor_id. | |
| Permite detectar menciones en texto libre. | |
| """ | |
| return { | |
| a["twitter"].lstrip("@").lower(): a["id"] | |
| for a in actores | |
| if a.get("twitter") | |
| } | |
| def _construir_mapa_nombres(actores: list[dict]) -> dict[str, str]: | |
| """ | |
| Construye un mapa apellido/apodo → actor_id para detectar menciones | |
| en texto libre. Excluye apellidos ambiguos (múltiples actores). | |
| """ | |
| from collections import defaultdict | |
| candidatos: dict[str, list[str]] = defaultdict(list) | |
| particulas = {"de", "del", "la", "el", "los", "las", "y", "e"} | |
| for actor in actores: | |
| palabras = actor["nombre"].split() | |
| for palabra in reversed(palabras): | |
| if palabra.lower() not in particulas and len(palabra) > 3: | |
| candidatos[palabra.lower()].append(actor["id"]) | |
| break | |
| for apodo in re.findall(r"'([^']+)'", actor["nombre"]): | |
| if len(apodo) > 2: | |
| candidatos[apodo.lower()].append(actor["id"]) | |
| mapa = {k: v[0] for k, v in candidatos.items() if len(v) == 1} | |
| mapa["kirchner"] = "kirchner_cristina" | |
| mapa["cfk"] = "kirchner_cristina" | |
| return mapa | |
| def _extraer_menciones( | |
| texto: str, | |
| handle_map: dict[str, str], | |
| nombre_map: dict[str, str] | None = None, | |
| ) -> list[str]: | |
| """ | |
| Extrae actor_ids mencionados en un texto via @handles y apellidos. | |
| Combina detección por @handle (tweets) y por apellido/apodo | |
| (artículos periodísticos que no usan @). | |
| """ | |
| texto_lower = texto.lower() | |
| encontrados: set[str] = set() | |
| for handle in re.findall(r"@(\w+)", texto_lower): | |
| if handle in handle_map: | |
| encontrados.add(handle_map[handle]) | |
| if nombre_map: | |
| for keyword, actor_id in nombre_map.items(): | |
| if re.search(r"\b" + re.escape(keyword) + r"\b", texto_lower): | |
| encontrados.add(actor_id) | |
| return list(encontrados) | |
| def _leer_posts_supabase(cluster: str | None = None, dias: int = 90) -> list[dict]: | |
| """ | |
| Lee posts de Supabase, opcionalmente filtrados por espectro ideológico. | |
| Parámetros: | |
| cluster: código de espectro ("ED", "CD", "C", "CI", "EI") o None para todos. | |
| dias: ventana temporal hacia atrás en días (default: 90). | |
| Devuelve lista de dicts con actor_id, titulo, resumen. | |
| """ | |
| if not SUPABASE_URL or not SUPABASE_KEY: | |
| return [] | |
| from supabase import create_client | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| corte = (datetime.now(timezone.utc) - timedelta(days=dias)).isoformat() | |
| if cluster: | |
| etiquetas = _cargar_etiquetas() | |
| ids_en_cluster = [ | |
| aid for aid, info in etiquetas.items() | |
| if info.get("espectro") == cluster | |
| ] | |
| if not ids_en_cluster: | |
| logger.warning(f"[influence_scorer] cluster '{cluster}' sin actores conocidos") | |
| return [] | |
| return _paginar( | |
| lambda: client.table("posts") | |
| .select("actor_id, titulo, resumen") | |
| .in_("actor_id", ids_en_cluster) | |
| .gte("publicado", corte) | |
| ) | |
| else: | |
| return _paginar( | |
| lambda: client.table("posts") | |
| .select("actor_id, titulo, resumen") | |
| .gte("publicado", corte) | |
| ) | |
| def _construir_grafo(posts: list[dict], actores: list[dict]) -> nx.DiGraph: | |
| """ | |
| Construye un grafo dirigido de influencia a partir de los posts recolectados. | |
| Arista A → B significa que el actor A mencionó al actor B en sus posts. | |
| El peso de la arista es el número de menciones. | |
| """ | |
| G = nx.DiGraph() | |
| etiquetas = _cargar_etiquetas() | |
| # Agregar todos los actores como nodos | |
| for actor in actores: | |
| aid = actor["id"] | |
| if not etiquetas.get(aid): | |
| logger.warning(f"[influence_scorer] actor '{aid}' sin etiqueta en ideology_labels.json — espectro='C' por defecto") | |
| G.add_node(aid, | |
| nombre=actor["nombre"], | |
| categoria=actor["categoria"], | |
| espectro=etiquetas.get(aid, {}).get("espectro", "C")) | |
| handle_map = _handles_a_ids(actores) | |
| nombre_map = _construir_mapa_nombres(actores) | |
| # Construir aristas a partir de menciones en posts | |
| for post in posts: | |
| origen = post.get("actor_id", "") | |
| if not origen or origen not in G: | |
| continue | |
| texto = f"{post.get('titulo', '')} {post.get('resumen', '')}" | |
| mencionados = _extraer_menciones(texto, handle_map, nombre_map) | |
| for destino in mencionados: | |
| if destino == origen or destino not in G: | |
| continue | |
| if G.has_edge(origen, destino): | |
| G[origen][destino]["weight"] += 1 | |
| else: | |
| G.add_edge(origen, destino, weight=1) | |
| # Filtrar aristas con peso mínimo | |
| aristas_debiles = [ | |
| (u, v) for u, v, d in G.edges(data=True) | |
| if d.get("weight", 0) < GRAPH_MIN_EDGE_WEIGHT | |
| ] | |
| G.remove_edges_from(aristas_debiles) | |
| logger.info( | |
| f"[influence_scorer] grafo: {G.number_of_nodes()} nodos, " | |
| f"{G.number_of_edges()} aristas" | |
| ) | |
| return G | |
| def _guardar_en_supabase(scores: dict[str, float], cluster: str | None) -> None: | |
| """ | |
| Guarda los scores de PageRank en la tabla influence_scores de Supabase. | |
| Crea la tabla si no existe via Management API. Hace upsert de los resultados. | |
| """ | |
| if not SUPABASE_URL or not SUPABASE_KEY: | |
| logger.warning("[influence_scorer] Supabase no configurado — scores no guardados") | |
| return | |
| # Crear tabla si no existe | |
| _crear_tabla_influence_scores() | |
| from supabase import create_client | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| ahora = datetime.now(timezone.utc).isoformat() | |
| rows = [ | |
| { | |
| "actor_id": actor_id, | |
| "pagerank_score": round(score, 8), | |
| "cluster": cluster or "global", | |
| "calculado_en": ahora, | |
| } | |
| for actor_id, score in scores.items() | |
| ] | |
| try: | |
| client.table("influence_scores").upsert(rows).execute() | |
| logger.info(f"[influence_scorer] {len(rows)} scores guardados en Supabase") | |
| except Exception as e: | |
| logger.error(f"[influence_scorer] error guardando en Supabase: {e}") | |
| def _crear_tabla_influence_scores() -> None: | |
| """Crea la tabla influence_scores en Supabase si no existe (solo una vez por proceso).""" | |
| global _TABLA_INFLUENCE_VERIFICADA | |
| if _TABLA_INFLUENCE_VERIFICADA: | |
| return | |
| _TABLA_INFLUENCE_VERIFICADA = True | |
| import os | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| token = os.getenv("SUPABASE_MANAGEMENT_TOKEN", "") | |
| if not token: | |
| logger.warning("[influence_scorer] sin SUPABASE_MANAGEMENT_TOKEN — tabla no creada") | |
| return | |
| import requests as req | |
| ref = SUPABASE_URL.replace("https://", "").replace(".supabase.co", "") | |
| sql = """ | |
| CREATE TABLE IF NOT EXISTS influence_scores ( | |
| actor_id TEXT NOT NULL, | |
| cluster TEXT NOT NULL DEFAULT 'global', | |
| pagerank_score FLOAT NOT NULL, | |
| calculado_en TIMESTAMPTZ NOT NULL DEFAULT NOW(), | |
| PRIMARY KEY (actor_id, cluster) | |
| ); | |
| ALTER TABLE influence_scores ENABLE ROW LEVEL SECURITY; | |
| DO $$ BEGIN | |
| IF NOT EXISTS (SELECT 1 FROM pg_policies WHERE tablename='influence_scores' AND policyname='anon_select_influence') THEN | |
| CREATE POLICY anon_select_influence ON influence_scores FOR SELECT TO anon USING (true); | |
| END IF; | |
| IF NOT EXISTS (SELECT 1 FROM pg_policies WHERE tablename='influence_scores' AND policyname='anon_insert_influence') THEN | |
| CREATE POLICY anon_insert_influence ON influence_scores FOR INSERT TO anon WITH CHECK (true); | |
| END IF; | |
| IF NOT EXISTS (SELECT 1 FROM pg_policies WHERE tablename='influence_scores' AND policyname='anon_update_influence') THEN | |
| CREATE POLICY anon_update_influence ON influence_scores FOR UPDATE TO anon USING (true); | |
| END IF; | |
| END $$; | |
| """ | |
| try: | |
| resp = req.post( | |
| f"https://api.supabase.com/v1/projects/{ref}/database/query", | |
| headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, | |
| json={"query": sql}, | |
| timeout=30, | |
| ) | |
| if resp.status_code in (200, 201): | |
| logger.info("[influence_scorer] tabla influence_scores lista") | |
| else: | |
| logger.warning(f"[influence_scorer] respuesta al crear tabla: {resp.status_code}") | |
| except Exception as e: | |
| logger.error(f"[influence_scorer] error creando tabla: {e}") | |
| # ── Score de influencia social compuesto ───────────────────────────────────── | |
| def calculate_social_influence() -> dict[str, dict]: | |
| """ | |
| Calcula el score de influencia social compuesto para los 65 actores. | |
| Fórmula: | |
| Para actores (politico/periodista/influencer): | |
| engagement_bruto = replies*4 + likes*3 + retweets*2 + views*1 | |
| (suma de todos sus posts en los últimos 90 días) | |
| Para medios (categoria == 'medio'): | |
| engagement_bruto = COUNT(menciones recibidas en posts) * 1000 | |
| (cuántos de los 65 actores citan sus artículos) | |
| Normalización: | |
| engagement_norm = engagement_bruto / max(engagement_bruto) | |
| pagerank_norm = pagerank_score / max(pagerank_score) | |
| Score final: | |
| influencia_social = engagement_norm*0.6 + pagerank_norm*0.4 | |
| Tamaño de nodo (escala logarítmica): | |
| node_size = 10 + log10(influencia_social * 1000 + 1) * 15 | |
| → mínimo 10, máximo ~55 | |
| Guarda resultados en tabla social_influence_scores de Supabase. | |
| Devuelve: | |
| Dict {actor_id: {influencia_social, engagement_bruto, engagement_norm, | |
| pagerank_norm, node_size, pagerank_score}} | |
| """ | |
| if not SUPABASE_URL or not SUPABASE_KEY: | |
| return {} | |
| from supabase import create_client | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| actores = _cargar_actores() | |
| actores_map = {a["id"]: a for a in actores} | |
| etiquetas = _cargar_etiquetas() | |
| # Leer PageRank guardado | |
| pr_resp = client.table("influence_scores").select("actor_id, pagerank_score").eq("cluster", "global").execute() | |
| pageranks = {r["actor_id"]: r["pagerank_score"] for r in (pr_resp.data or [])} | |
| # Leer posts últimos 90 días con métricas de engagement (paginado). | |
| # Solo tweets propios (NULL) y replies enviados por el actor. | |
| # Excluidos: retweets y reply_recibido — sus contadores pertenecen a terceros. | |
| corte = (datetime.now(timezone.utc) - timedelta(days=90)).isoformat() | |
| posts = _paginar( | |
| lambda: client.table("posts").select( | |
| "actor_id, reply_count, like_count, retweet_count, view_count" | |
| ).gte("publicado", corte).or_("tipo_interaccion.is.null,tipo_interaccion.eq.reply") | |
| ) | |
| # Calcular engagement_bruto para todos (medios y actores usan la misma fórmula). | |
| # Para medios: el engagement viene de sus tweets (reply*4 + like*3 + RT*2 + view*1) | |
| # igual que para políticos/periodistas/influencers. | |
| engagement_actores: dict[str, int] = {a["id"]: 0 for a in actores} | |
| for p in posts: | |
| aid = p.get("actor_id", "") | |
| if aid not in engagement_actores: | |
| continue | |
| engagement_actores[aid] += ( | |
| p.get("reply_count", 0) * 4 | |
| + p.get("like_count", 0) * 3 | |
| + p.get("retweet_count", 0) * 2 | |
| + p.get("view_count", 0) * 1 | |
| ) | |
| # Normalizar engagement y pagerank | |
| max_eng = max(engagement_actores.values(), default=1) or 1 | |
| max_pr = max(pageranks.values(), default=1) or 1 | |
| resultados: dict[str, dict] = {} | |
| ahora = datetime.now(timezone.utc).isoformat() | |
| for actor in actores: | |
| aid = actor["id"] | |
| eng_bruto = engagement_actores.get(aid, 0) | |
| pr_score = pageranks.get(aid, 0.0) | |
| eng_norm = eng_bruto / max_eng | |
| pr_norm = pr_score / max_pr | |
| influencia = eng_norm * 0.6 + pr_norm * 0.4 | |
| node_size = 10 + math.log10(influencia * 1000 + 1) * 15 | |
| resultados[aid] = { | |
| "influencia_social": round(influencia, 6), | |
| "engagement_bruto": eng_bruto, | |
| "engagement_norm": round(eng_norm, 6), | |
| "pagerank_norm": round(pr_norm, 6), | |
| "pagerank_score": pr_score, | |
| "node_size": round(node_size, 2), | |
| } | |
| # Guardar en Supabase | |
| try: | |
| rows = [ | |
| { | |
| "actor_id": aid, | |
| "influencia_social": v["influencia_social"], | |
| "engagement_bruto": v["engagement_bruto"], | |
| "engagement_norm": v["engagement_norm"], | |
| "pagerank_norm": v["pagerank_norm"], | |
| "node_size": v["node_size"], | |
| "calculado_en": ahora, | |
| } | |
| for aid, v in resultados.items() | |
| ] | |
| client.table("social_influence_scores").upsert(rows, on_conflict="actor_id").execute() | |
| logger.info(f"[influence_scorer] social_influence calculado para {len(rows)} actores") | |
| except Exception as e: | |
| logger.error(f"[influence_scorer] error guardando social_influence: {e}") | |
| return resultados | |
| def get_social_influence_ranking(top_n: int = 10, cluster: str | None = None) -> list[dict]: | |
| """ | |
| Retorna el top N de actores por influencia_social compuesta. | |
| Parámetros: | |
| top_n: cuántos actores incluir (default: 10) | |
| cluster: filtrar por espectro ideológico antes de rankear | |
| Cada item devuelve: | |
| actor_id, nombre, espectro, categoria, influencia_social, | |
| engagement_bruto, pagerank_score, node_size | |
| """ | |
| scores = calculate_social_influence() | |
| if not scores: | |
| return [] | |
| actores = {a["id"]: a for a in _cargar_actores()} | |
| etiquetas = _cargar_etiquetas() | |
| items = [ | |
| { | |
| "actor_id": aid, | |
| "nombre": actores.get(aid, {}).get("nombre", aid), | |
| "espectro": etiquetas.get(aid, {}).get("espectro", "?"), | |
| "categoria": actores.get(aid, {}).get("categoria", "?"), | |
| "influencia_social": v["influencia_social"], | |
| "engagement_bruto": v["engagement_bruto"], | |
| "pagerank_score": v["pagerank_score"], | |
| "node_size": v["node_size"], | |
| } | |
| for aid, v in scores.items() | |
| ] | |
| if cluster: | |
| items = [i for i in items if i["espectro"] == cluster] | |
| items.sort(key=lambda x: x["influencia_social"], reverse=True) | |
| return items[:top_n] | |
| # ── Funciones públicas ──────────────────────────────────────────────────────── | |
| def calculate_pagerank(cluster: str | None = None) -> dict[str, float]: | |
| """ | |
| Calcula el PageRank de todos los actores a partir de sus menciones cruzadas. | |
| Parámetros: | |
| cluster: filtrar por espectro ideológico ("ED", "CD", "C", "CI", "EI") | |
| o None para calcular sobre todos los actores. | |
| Devuelve: | |
| Dict {actor_id: pagerank_score}. Dict vacío si no hay posts en Supabase. | |
| """ | |
| actores = _cargar_actores() | |
| try: | |
| posts = _leer_posts_supabase(cluster) | |
| except Exception as e: | |
| logger.error(f"[influence_scorer] error leyendo Supabase: {e}") | |
| return {} | |
| if not posts: | |
| logger.info("[influence_scorer] sin posts en Supabase — PageRank no calculado") | |
| return {} | |
| G = _construir_grafo(posts, actores) | |
| # Si el grafo no tiene aristas, PageRank es uniforme | |
| if G.number_of_edges() == 0: | |
| logger.info("[influence_scorer] grafo sin aristas — distribución uniforme") | |
| n = G.number_of_nodes() | |
| return {node: 1.0 / n for node in G.nodes()} if n > 0 else {} | |
| scores = nx.pagerank(G, alpha=GRAPH_PAGERANK_ALPHA, weight="weight") | |
| logger.info( | |
| f"[influence_scorer] PageRank calculado — " | |
| f"cluster={cluster or 'global'}, max_score={max(scores.values()):.6f}" | |
| ) | |
| _guardar_en_supabase(scores, cluster) | |
| return scores | |
| def get_actors_by_tag(tag: str) -> list[dict]: | |
| """ | |
| Retorna todos los actores que tienen el soft_tag dado, con su espectro y PageRank. | |
| Devuelve lista de dicts: {actor_id, nombre, espectro, soft_tags, pagerank_score} | |
| ordenada por pagerank_score descendente. | |
| """ | |
| etiquetas = _cargar_etiquetas() | |
| actores = {a["id"]: a for a in _cargar_actores()} | |
| # Leer scores actuales de Supabase | |
| scores: dict[str, float] = {} | |
| if SUPABASE_URL and SUPABASE_KEY: | |
| try: | |
| from supabase import create_client | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| resp = client.table("influence_scores").select("actor_id, pagerank_score").eq("cluster", "global").execute() | |
| scores = {r["actor_id"]: r["pagerank_score"] for r in (resp.data or [])} | |
| except Exception as e: | |
| logger.warning(f"[influence_scorer] no se pudieron leer scores: {e}") | |
| resultado = [] | |
| for actor_id, info in etiquetas.items(): | |
| if tag in info.get("soft_tags", []): | |
| actor = actores.get(actor_id, {}) | |
| resultado.append({ | |
| "actor_id": actor_id, | |
| "nombre": actor.get("nombre", actor_id), | |
| "espectro": info.get("espectro", "?"), | |
| "soft_tags": info.get("soft_tags", []), | |
| "pagerank_score": scores.get(actor_id, 0.0), | |
| }) | |
| return sorted(resultado, key=lambda x: x["pagerank_score"], reverse=True) | |
| def get_tag_clusters() -> dict: | |
| """ | |
| Para cada soft_tag retorna: cuántos actores lo tienen, qué espectros están | |
| representados, y el actor más influyente (mayor PageRank). | |
| Replica la vista 'Channel Landscape' de Recfluence adaptada a Argentina. | |
| Devuelve dict: {tag: {count, espectros, top_actor}} | |
| """ | |
| etiquetas = _cargar_etiquetas() | |
| actores = {a["id"]: a for a in _cargar_actores()} | |
| scores: dict[str, float] = {} | |
| if SUPABASE_URL and SUPABASE_KEY: | |
| try: | |
| from supabase import create_client | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| resp = client.table("influence_scores").select("actor_id, pagerank_score").eq("cluster", "global").execute() | |
| scores = {r["actor_id"]: r["pagerank_score"] for r in (resp.data or [])} | |
| except Exception as e: | |
| logger.warning(f"[influence_scorer] no se pudieron leer scores: {e}") | |
| from collections import defaultdict | |
| clusters: dict[str, dict] = defaultdict(lambda: {"count": 0, "espectros": set(), "actores": []}) | |
| for actor_id, info in etiquetas.items(): | |
| actor = actores.get(actor_id, {}) | |
| score = scores.get(actor_id, 0.0) | |
| for tag in info.get("soft_tags", []): | |
| clusters[tag]["count"] += 1 | |
| clusters[tag]["espectros"].add(info.get("espectro", "?")) | |
| clusters[tag]["actores"].append({"actor_id": actor_id, "nombre": actor.get("nombre", actor_id), "score": score}) | |
| resultado = {} | |
| for tag, data in sorted(clusters.items()): | |
| top = max(data["actores"], key=lambda x: x["score"], default=None) | |
| resultado[tag] = { | |
| "count": data["count"], | |
| "espectros": sorted(data["espectros"]), | |
| "top_actor": top["nombre"] if top else None, | |
| "top_score": round(top["score"], 6) if top else 0.0, | |
| } | |
| return resultado | |
| def get_influence_ranking_from_db(top_n: int = 10, cluster: str | None = None) -> list[dict]: | |
| """ | |
| Lee el ranking de PageRank directamente desde Supabase sin recalcular. | |
| Úsalo en el dashboard — el pipeline ya calculó y guardó los scores. | |
| Solo llama a calculate_pagerank() si no hay datos en Supabase. | |
| """ | |
| if not SUPABASE_URL or not SUPABASE_KEY: | |
| return [] | |
| try: | |
| from supabase import create_client | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| query = client.table("influence_scores").select("actor_id, pagerank_score").eq("cluster", "global") | |
| resp = query.execute() | |
| scores = {r["actor_id"]: r["pagerank_score"] for r in (resp.data or [])} | |
| except Exception as e: | |
| logger.warning(f"[influence_scorer] error leyendo ranking de DB: {e}") | |
| return [] | |
| if not scores: | |
| logger.info("[influence_scorer] sin scores en DB, calculando on-demand") | |
| return get_influence_ranking(top_n=top_n, cluster=cluster) | |
| actores = {a["id"]: a for a in _cargar_actores()} | |
| etiquetas = _cargar_etiquetas() | |
| items = [ | |
| { | |
| "posicion": 0, | |
| "actor_id": aid, | |
| "nombre": actores.get(aid, {}).get("nombre", aid), | |
| "espectro": etiquetas.get(aid, {}).get("espectro", "?"), | |
| "score": round(score, 6), | |
| } | |
| for aid, score in scores.items() | |
| ] | |
| if cluster: | |
| items = [i for i in items if i["espectro"] == cluster] | |
| items.sort(key=lambda x: x["score"], reverse=True) | |
| for i, item in enumerate(items[:top_n]): | |
| item["posicion"] = i + 1 | |
| return items[:top_n] | |
| def get_social_influence_ranking_from_db(top_n: int = 10, cluster: str | None = None) -> list[dict]: | |
| """ | |
| Lee el ranking de influencia social directamente desde Supabase sin recalcular. | |
| Úsalo en el dashboard — el pipeline ya calculó y guardó los scores. | |
| Solo llama a calculate_social_influence() si no hay datos en Supabase. | |
| """ | |
| if not SUPABASE_URL or not SUPABASE_KEY: | |
| return [] | |
| try: | |
| from supabase import create_client | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| resp = client.table("social_influence_scores").select( | |
| "actor_id, influencia_social, engagement_bruto, node_size" | |
| ).execute() | |
| rows = resp.data or [] | |
| except Exception as e: | |
| logger.warning(f"[influence_scorer] error leyendo social ranking de DB: {e}") | |
| return [] | |
| if not rows: | |
| logger.info("[influence_scorer] sin social scores en DB, calculando on-demand") | |
| return get_social_influence_ranking(top_n=top_n, cluster=cluster) | |
| actores = {a["id"]: a for a in _cargar_actores()} | |
| etiquetas = _cargar_etiquetas() | |
| pageranks = {} | |
| try: | |
| from supabase import create_client | |
| client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) | |
| pr_resp = client.table("influence_scores").select("actor_id, pagerank_score").eq("cluster", "global").execute() | |
| pageranks = {r["actor_id"]: r["pagerank_score"] for r in (pr_resp.data or [])} | |
| except Exception: | |
| pass | |
| items = [] | |
| for r in rows: | |
| aid = r["actor_id"] | |
| espectro = etiquetas.get(aid, {}).get("espectro", "?") | |
| if cluster and espectro != cluster: | |
| continue | |
| items.append({ | |
| "posicion": 0, | |
| "actor_id": aid, | |
| "nombre": actores.get(aid, {}).get("nombre", aid), | |
| "espectro": espectro, | |
| "categoria": actores.get(aid, {}).get("categoria", "?"), | |
| "influencia_social": r.get("influencia_social", 0.0), | |
| "engagement_bruto": r.get("engagement_bruto", 0), | |
| "pagerank_score": pageranks.get(aid, 0.0), | |
| "node_size": r.get("node_size", 10.0), | |
| }) | |
| items.sort(key=lambda x: x["influencia_social"], reverse=True) | |
| for i, item in enumerate(items[:top_n]): | |
| item["posicion"] = i + 1 | |
| return items[:top_n] | |
| def get_influence_ranking(top_n: int = 10, cluster: str | None = None) -> list[dict]: | |
| """ | |
| Devuelve el ranking de actores más influyentes según PageRank. | |
| Parámetros: | |
| top_n: cuántos actores incluir en el ranking (default: 10) | |
| cluster: filtrar por espectro ideológico o None para todos | |
| Devuelve: | |
| Lista de dicts con: posicion, actor_id, nombre, espectro, score. | |
| Lista vacía si no hay datos. | |
| """ | |
| scores = calculate_pagerank(cluster) | |
| if not scores: | |
| return [] | |
| actores = {a["id"]: a for a in _cargar_actores()} | |
| etiquetas = _cargar_etiquetas() | |
| ranking_ordenado = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_n] | |
| return [ | |
| { | |
| "posicion": i + 1, | |
| "actor_id": actor_id, | |
| "nombre": actores.get(actor_id, {}).get("nombre", actor_id), | |
| "espectro": etiquetas.get(actor_id, {}).get("espectro", "?"), | |
| "score": round(score, 6), | |
| } | |
| for i, (actor_id, score) in enumerate(ranking_ordenado) | |
| ] | |