arginfluence / processing /narrative_generator.py
successjourney's picture
Deploy ArgInfluence dashboard
e3408e0 verified
Raw History Blame Contribute Delete
24.2 kB
"""
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 []