""" narrative_generator.py — Genera observaciones automáticas en lenguaje natural. Lee actor_scores_history y actor_topics de Supabase, detecta los 3 actores con mayor cambio de PageRank en el período, y genera una observación de 3 oraciones por actor via Llama-3.2-1B-Instruct (HF Inference API). Función pública principal: generar_narrativas(dias=30) -> list[dict] """ import json import logging import time from datetime import datetime, timezone, timedelta from pathlib import Path import sys import requests sys.path.insert(0, str(Path(__file__).parent.parent)) from config import ( HUGGINGFACE_API_KEY, HF_API_URL_NARRATIVE, HF_RETRY_DELAY, HF_RATELIMIT_DELAY, SUPABASE_URL, SUPABASE_KEY, SUPABASE_SERVICE_KEY, SEED_ACTORS_PATH, LOG_FORMAT, LOG_DATE_FORMAT, LOG_LEVEL, ) from utils.db import get_client, paginar logging.basicConfig(format=LOG_FORMAT, datefmt=LOG_DATE_FORMAT) logger = logging.getLogger("narrative_generator") logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO)) # ── Helpers de datos ─────────────────────────────────────────────────────────── def _cargar_nombres_actores() -> dict[str, str]: """Devuelve dict {actor_id: nombre} desde seed_actors.json.""" with open(SEED_ACTORS_PATH, encoding="utf-8") as f: return {a["id"]: a["nombre"] for a in json.load(f)["actors"]} def _leer_historial_periodo(dias: int) -> dict[str, list[dict]]: """ Lee actor_scores_history para el período indicado. Devuelve dict {actor_id: [rows ordenadas por timestamp asc]}. """ if not SUPABASE_URL or not SUPABASE_KEY: return {} from supabase import create_client client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) from_date = (datetime.now(timezone.utc) - timedelta(days=dias)).isoformat() historial: dict[str, list[dict]] = {} offset, PAGE = 0, 1000 while True: resp = ( client.table("actor_scores_history") .select("actor_id, timestamp, pagerank, engagement_total, sentimiento_promedio") .gte("timestamp", from_date) .order("timestamp") .range(offset, offset + PAGE - 1) .execute() ) batch = resp.data or [] for row in batch: historial.setdefault(row["actor_id"], []).append(row) if len(batch) < PAGE: break offset += PAGE return historial def _leer_topicos_dominantes(actor_ids: list[str]) -> dict[str, str]: """ Lee el tópico con posicion=1 (dominante) para cada actor desde actor_topics. Devuelve dict {actor_id: tema}. Actores sin datos quedan fuera del dict. """ if not actor_ids or 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("actor_topics") .select("actor_id, tema") .in_("actor_id", actor_ids) .eq("posicion", 1) .execute() ) return {row["actor_id"]: row["tema"] for row in (resp.data or [])} except Exception as e: logger.warning(f"[narrative] no se pudieron leer tópicos: {e}") return {} # ── Cálculo de deltas ────────────────────────────────────────────────────────── def _calcular_deltas(historial: dict[str, list[dict]]) -> dict[str, dict]: """ Calcula deltas de PageRank, engagement y sentimiento para cada actor. Requiere al menos 2 snapshots — actores con solo 1 se omiten. Devuelve dict {actor_id: {pr_inicio, pr_fin, delta_pr_pct, eng_inicio, eng_fin, delta_eng_pct, sent_inicio, sent_fin, delta_sent, n_snapshots}}. """ deltas = {} for actor_id, rows in historial.items(): if len(rows) < 2: continue primero = rows[0] ultimo = rows[-1] pr_i = primero.get("pagerank") or 0.0 pr_f = ultimo.get("pagerank") or 0.0 eng_i = float(primero.get("engagement_total") or 0.0) eng_f = float(ultimo.get("engagement_total") or 0.0) sent_i = primero.get("sentimiento_promedio") sent_f = ultimo.get("sentimiento_promedio") delta_pr_pct = ((pr_f - pr_i) / pr_i * 100) if pr_i > 0 else 0.0 delta_eng_pct = ((eng_f - eng_i) / eng_i * 100) if eng_i > 0 else 0.0 delta_sent = ( round(sent_f - sent_i, 3) if (sent_i is not None and sent_f is not None) else None ) deltas[actor_id] = { "pr_inicio": pr_i, "pr_fin": pr_f, "delta_pr_pct": round(delta_pr_pct, 1), "eng_inicio": eng_i, "eng_fin": eng_f, "delta_eng_pct": round(delta_eng_pct, 1), "sent_inicio": sent_i, "sent_fin": sent_f, "delta_sent": delta_sent, "n_snapshots": len(rows), } return deltas def _seleccionar_top_actores(deltas: dict[str, dict], n: int = 3) -> list[str]: """Devuelve los n actor_ids con mayor cambio absoluto de PageRank.""" return sorted( deltas.keys(), key=lambda aid: abs(deltas[aid]["delta_pr_pct"]), reverse=True, )[:n] # ── Cambios relacionales ─────────────────────────────────────────────────────── def _leer_posts_interacciones(from_date_anterior: str) -> list[dict]: """ Lee posts con interaccion_con no nulo desde from_date_anterior en adelante. Usa paginación vía utils.db.paginar para no agotar la memoria con datasets grandes. """ client = get_client() return paginar( lambda: ( client.table("posts") .select("actor_id, publicado, interaccion_con, tipo_interaccion") .gte("publicado", from_date_anterior) .not_.is_("interaccion_con", "null") ) ) def _construir_pares(posts: list[dict], actores_validos: set[str]) -> dict[tuple, dict]: """ Construye dict {(actor_a, actor_b): {"rt": n, "reply": m}} desde una lista de posts. Solo incluye pares donde el receptor (interaccion_con) está en actores_validos. """ pares: dict[tuple, dict] = {} for post in posts: origen = post.get("actor_id") destino = post.get("interaccion_con") tipo = (post.get("tipo_interaccion") or "").lower() if not origen or not destino: continue if destino not in actores_validos: continue clave = (origen, destino) if clave not in pares: pares[clave] = {"rt": 0, "reply": 0} if tipo == "retweet": pares[clave]["rt"] += 1 elif tipo == "reply": pares[clave]["reply"] += 1 return pares def detectar_cambios_relacionales(dias: int = 7) -> dict: """ Compara las interacciones (RTs y replies) de los últimos `dias` días vs. el período anterior del mismo largo. Devuelve dict con: nuevas — pares que aparecen en el período reciente pero no en el anterior perdidas — pares que estaban en el anterior pero no en el reciente intensificadas — pares cuyo conteo total subió más del 50% debilitadas — pares cuyo conteo total bajó más del 50% resumen — texto legible con los conteos periodo_dias — valor de `dias` recibido """ vacio = { "nuevas": [], "perdidas": [], "intensificadas": [], "debilitadas": [], "resumen": "Sin cambios relacionales detectados", "periodo_dias": dias, } if not SUPABASE_URL or not SUPABASE_KEY: logger.warning("[relacional] Supabase no configurado — sin análisis relacional") return vacio try: ahora = datetime.now(timezone.utc) inicio_reciente = ahora - timedelta(days=dias) inicio_anterior = ahora - timedelta(days=dias * 2) # Una sola lectura que cubre ambas ventanas from_date_str = inicio_anterior.isoformat() posts_raw = _leer_posts_interacciones(from_date_str) if not posts_raw: logger.info("[relacional] sin posts con interacciones en el período doble") return vacio # Cargar actores válidos para filtrar receptores no mapeados nombres = _cargar_nombres_actores() actores_validos = set(nombres.keys()) # Separar las dos ventanas temporales posts_recientes = [] posts_anteriores = [] for post in posts_raw: publicado_str = post.get("publicado") or "" try: # Supabase devuelve TIMESTAMPTZ como string ISO pub = datetime.fromisoformat(publicado_str.replace("Z", "+00:00")) except (ValueError, AttributeError) as e: logger.debug(f"[relacional] fecha inválida en post, omitiendo: {e}") continue if pub >= inicio_reciente: posts_recientes.append(post) elif pub >= inicio_anterior: posts_anteriores.append(post) if not posts_recientes and not posts_anteriores: logger.info("[relacional] sin posts válidos en ninguna ventana") return vacio pares_recientes = _construir_pares(posts_recientes, actores_validos) pares_anteriores = _construir_pares(posts_anteriores, actores_validos) claves_recientes = set(pares_recientes.keys()) claves_anteriores = set(pares_anteriores.keys()) nuevas = [] perdidas = [] intensificadas = [] debilitadas = [] # Relaciones nuevas: existen en reciente pero no en anterior for clave in claves_recientes - claves_anteriores: actor_a, actor_b = clave conteos = pares_recientes[clave] total = conteos["rt"] + conteos["reply"] tipo_str = ( "retweet" if conteos["rt"] >= conteos["reply"] else "reply" ) nuevas.append({ "de": actor_a, "hacia": actor_b, "de_nombre": nombres.get(actor_a, actor_a), "hacia_nombre": nombres.get(actor_b, actor_b), "tipo": tipo_str, "n": total, }) # Relaciones perdidas: existían en anterior pero no en reciente for clave in claves_anteriores - claves_recientes: actor_a, actor_b = clave perdidas.append({ "de": actor_a, "hacia": actor_b, "de_nombre": nombres.get(actor_a, actor_a), "hacia_nombre": nombres.get(actor_b, actor_b), }) # Relaciones que persisten: comparar magnitud for clave in claves_recientes & claves_anteriores: actor_a, actor_b = clave antes = pares_anteriores[clave]["rt"] + pares_anteriores[clave]["reply"] ahora_cnt = pares_recientes[clave]["rt"] + pares_recientes[clave]["reply"] if antes == 0: continue cambio_pct = (ahora_cnt - antes) / antes * 100 entrada = { "de": actor_a, "hacia": actor_b, "de_nombre": nombres.get(actor_a, actor_a), "hacia_nombre": nombres.get(actor_b, actor_b), "antes": antes, "ahora": ahora_cnt, } if cambio_pct > 50: intensificadas.append(entrada) elif cambio_pct < -50: debilitadas.append(entrada) # Ordenar por relevancia (mayor volumen primero) nuevas.sort(key=lambda x: x["n"], reverse=True) intensificadas.sort(key=lambda x: x["ahora"], reverse=True) debilitadas.sort(key=lambda x: x["antes"], reverse=True) # Armar texto de resumen partes = [] if nuevas: partes.append(f"{len(nuevas)} relación{'es' if len(nuevas) > 1 else ''} nueva{'s' if len(nuevas) > 1 else ''}") if perdidas: partes.append(f"{len(perdidas)} perdida{'s' if len(perdidas) > 1 else ''}") if intensificadas: partes.append(f"{len(intensificadas)} intensificada{'s' if len(intensificadas) > 1 else ''}") if debilitadas: partes.append(f"{len(debilitadas)} debilitada{'s' if len(debilitadas) > 1 else ''}") resumen = ", ".join(partes) if partes else "Sin cambios relacionales detectados" logger.info(f"[relacional] período {dias}d — {resumen}") return { "nuevas": nuevas, "perdidas": perdidas, "intensificadas": intensificadas, "debilitadas": debilitadas, "resumen": resumen, "periodo_dias": dias, } except Exception as e: logger.error(f"[relacional] error en detectar_cambios_relacionales: {e}") return vacio def _contexto_relacional_para_actor( actor_id: str, cambios: dict, dias: int, ) -> str | None: """ Si el actor aparece en cambios relacionales (nuevas o perdidas), devuelve un párrafo descriptivo para enriquecer el prompt del LLM. Devuelve None si el actor no tiene cambios relacionales relevantes. """ fragmentos = [] # Relaciones nuevas donde el actor es origen nuevas_como_origen = [ r for r in cambios.get("nuevas", []) if r["de"] == actor_id ] for rel in nuevas_como_origen[:2]: # Limitar a 2 para no inflar el prompt tipo_label = "RTs" if rel["tipo"] == "retweet" else "replies" fragmentos.append( f"empezó a interactuar con {rel['hacia_nombre']} " f"({rel['n']} {tipo_label} en los últimos {dias} días)" ) # Relaciones perdidas donde el actor es origen perdidas_como_origen = [ r for r in cambios.get("perdidas", []) if r["de"] == actor_id ] for rel in perdidas_como_origen[:2]: fragmentos.append(f"dejó de interactuar con {rel['hacia_nombre']}") if not fragmentos: return None return " y ".join(fragmentos) # ── Generación de texto ──────────────────────────────────────────────────────── def _construir_narrativa_fallback( nombre: str, delta_pr: float, delta_eng: float, topico: str, periodo_label: str, ) -> str: """ Genera observación sin LLM usando la plantilla del spec. Se usa cuando el LLM no responde o devuelve texto inválido. """ dir_pr = "subió" if delta_pr >= 0 else "bajó" dir_eng = "subió" if delta_eng >= 0 else "bajó" return ( f"{nombre} {dir_pr} {abs(delta_pr):.1f}% en PageRank durante los últimos {periodo_label}. " f"En ese tiempo habló principalmente de {topico}. " f"Su engagement {dir_eng} {abs(delta_eng):.1f}% en el mismo período." ) def _llamar_llm(prompt_usuario: str) -> str | None: """ Llama a Llama-3.2-1B-Instruct via HF router (chat completions, formato OpenAI). Devuelve el texto generado o None si falla. """ if not HUGGINGFACE_API_KEY: logger.warning("[narrative] HUGGINGFACE_API_KEY no configurada — sin generación LLM") return None payload = { "model": "meta-llama/Llama-3.2-1B-Instruct", "messages": [ { "role": "system", "content": ( "Sos un analista político argentino conciso. " "Respondés ÚNICAMENTE en español rioplatense. " "No agregás texto fuera del formato solicitado." ), }, {"role": "user", "content": prompt_usuario}, ], "max_tokens": 200, "temperature": 0.2, "stream": False, } headers = { "Authorization": f"Bearer {HUGGINGFACE_API_KEY}", "Content-Type": "application/json", } for intento in range(1, 3): try: resp = requests.post( HF_API_URL_NARRATIVE, headers=headers, json=payload, timeout=45, ) if resp.status_code == 200: data = resp.json() return data["choices"][0]["message"]["content"].strip() elif resp.status_code == 503: logger.warning( f"[narrative] 503 modelo cargando (intento {intento}) — " f"esperando {HF_RETRY_DELAY}s" ) time.sleep(HF_RETRY_DELAY) elif resp.status_code == 429: logger.warning(f"[narrative] 429 rate limit — esperando {HF_RATELIMIT_DELAY}s") time.sleep(HF_RATELIMIT_DELAY) else: logger.error(f"[narrative] error {resp.status_code}: {resp.text[:200]}") return None except requests.exceptions.Timeout: logger.warning(f"[narrative] timeout en intento {intento}") if intento < 2: time.sleep(HF_RETRY_DELAY) except Exception as e: logger.error(f"[narrative] error inesperado llamando LLM: {e}") return None return None def _sanitizar_para_prompt(texto: str, max_chars: int = 80) -> str: """ Elimina caracteres que podrían alterar la estructura del prompt LLM. Trunca a max_chars para limitar el tamaño del prompt. """ # Reemplazar saltos de línea y caracteres de control que rompen el formato limpio = texto.replace("\n", " ").replace("\r", " ").replace('"', "'") # Remover secuencias que intentan inyectar roles o instrucciones for patron in ["system:", "user:", "assistant:", "```", "###", "IGNORE", "ignore"]: limpio = limpio.replace(patron, "") return limpio.strip()[:max_chars] def _generar_observacion( nombre: str, datos: dict, topico: str, dias: int, contexto_relacional: str | None = None, ) -> str: """ Genera la observación de 3 oraciones para un actor. Si hay contexto relacional (cambios de alianza), se usa como foco del prompt. Si no, usa métricas de PageRank/engagement. Fallback a plantilla si LLM falla. """ periodo_label = f"{dias} días" delta_pr = datos["delta_pr_pct"] delta_eng = datos["delta_eng_pct"] # Sanitizar campos que van al prompt para evitar inyección nombre_safe = _sanitizar_para_prompt(nombre, max_chars=60) topico_safe = _sanitizar_para_prompt(topico, max_chars=60) # Descripción de sentimiento para el prompt sent_str = "sin datos disponibles" if datos.get("delta_sent") is not None: dir_s = "mejoró" if datos["delta_sent"] >= 0 else "empeoró" sent_str = f"{dir_s} {abs(datos['delta_sent']):.2f} puntos (de {datos['sent_inicio']:+.2f} a {datos['sent_fin']:+.2f})" if contexto_relacional: # Prompt orientado a cambios de alianza/interacción — más interesante que métricas puras relacional_safe = _sanitizar_para_prompt(contexto_relacional, max_chars=200) prompt = ( f"Generá exactamente 3 oraciones analizando este cambio en las relaciones " f"del actor político argentino {nombre_safe}.\n\n" f"Cambio relacional detectado: {nombre_safe} {relacional_safe}.\n\n" f"Datos de contexto ({periodo_label}):\n" f"- PageRank: {datos['pr_inicio']:.5f} → {datos['pr_fin']:.5f} ({delta_pr:+.1f}%)\n" f"- Engagement: {datos['eng_inicio']:.0f} → {datos['eng_fin']:.0f} ({delta_eng:+.1f}%)\n" f"- Tópico dominante: {topico_safe}\n\n" f"Respondé SOLO con las 3 oraciones. Sin títulos. Sin texto adicional." ) else: prompt = ( f"Generá exactamente 3 oraciones sobre este actor político argentino. " f"Seguí el formato EXACTO indicado:\n\n" f"Formato: \"{nombre_safe} [subió/bajó] X% en PageRank durante los últimos {periodo_label}. " f"En ese tiempo habló principalmente de [tópico]. [Oración interpretativa breve basada en los datos].\"\n\n" f"Datos del período ({periodo_label}):\n" f"- Nombre: {nombre_safe}\n" f"- PageRank: {datos['pr_inicio']:.5f} → {datos['pr_fin']:.5f} ({delta_pr:+.1f}%)\n" f"- Engagement: {datos['eng_inicio']:.0f} → {datos['eng_fin']:.0f} ({delta_eng:+.1f}%)\n" f"- Sentimiento de audiencia: {sent_str}\n" f"- Tópico dominante: {topico_safe}\n\n" f"Respondé SOLO con las 3 oraciones. Sin títulos. Sin texto adicional." ) texto_llm = _llamar_llm(prompt) if texto_llm and len(texto_llm) > 40: logger.info(f"[narrative] observación LLM generada para {nombre}") return texto_llm logger.info(f"[narrative] usando fallback para {nombre}") return _construir_narrativa_fallback(nombre, delta_pr, delta_eng, topico, periodo_label) # ── Función pública ──────────────────────────────────────────────────────────── def generar_narrativas(dias: int = 30) -> list[dict]: """ Genera observaciones para los 3 actores con mayor cambio en el período. Parámetros: dias: período de análisis (7, 30 o 90) Devuelve: Lista de hasta 3 dicts con claves: actor_id (str) nombre (str) observacion (str) — 3 oraciones generadas topico (str) datos (dict) — {pr_inicio, pr_fin, delta_pr_pct, eng_inicio, eng_fin, delta_eng_pct, sent_inicio, sent_fin, delta_sent, n_snapshots} Lista vacía si no hay actores con ≥ 2 snapshots en el período. """ try: nombres = _cargar_nombres_actores() historial = _leer_historial_periodo(dias) if not historial: logger.info("[narrative] sin datos en actor_scores_history para el período") return [] deltas = _calcular_deltas(historial) if not deltas: logger.info("[narrative] ningún actor con ≥ 2 snapshots en el período") return [] top_ids = _seleccionar_top_actores(deltas, n=3) topicos = _leer_topicos_dominantes(top_ids) # Detectar cambios relacionales para enriquecer los prompts try: cambios_relacionales = detectar_cambios_relacionales(dias=min(dias, 30)) except Exception as e: logger.warning(f"[narrative] no se pudo obtener cambios relacionales: {e}") cambios_relacionales = {} resultado = [] for actor_id in top_ids: nombre = nombres.get(actor_id, actor_id) datos = deltas[actor_id] topico = topicos.get(actor_id, "varios temas") # Contexto relacional específico para este actor (None si no tiene cambios) ctx_relacional = _contexto_relacional_para_actor( actor_id, cambios_relacionales, dias ) if cambios_relacionales else None observacion = _generar_observacion(nombre, datos, topico, dias, ctx_relacional) resultado.append({ "actor_id": actor_id, "nombre": nombre, "observacion": observacion, "topico": topico, "datos": datos, }) logger.info(f"[narrative] procesado: {nombre}") return resultado except Exception as e: logger.error(f"[narrative] error en generar_narrativas: {e}") return []