Spaces:

successjourney
/

arginfluence

Running

arginfluence / processing /narrative_generator.py

Deploy ArgInfluence dashboard

e3408e0 verified 7 months ago

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 []

	"""
	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 []