Blame for processing/narrative_generator.py

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