osint_db/server/derived.py

"""Reconstrucción de las tablas derivadas (schema derived).

Regla dura: las derivadas contienen SOLO datos computados — ninguna lleva
columna que referencie notas (note_path prohibido aquí). Se reconstruyen
completas (DROP + CREATE) en cada ingest, así que su contenido siempre refleja
el último estado de las maestras.
"""

from __future__ import annotations

import json
import re

# Nombres de las derivadas que este módulo gestiona (para reportar en ingest).
DERIVED_TABLES = (
    "person_stats",
    "event_monthly",
    "contact_link_quality",
    "org_contacts",
)

# Relación persona -> organización en el frontmatter: "[[org-slug]] — rol".
_RELACION_RE = re.compile(r"\[\[([^\]]+)\]\]\s*(?:—|-|–)?\s*(.*)")


def rebuild_derived(conn) -> list:
    """Reconstruye todas las tablas derivadas sobre la conexión de escritura.

    Devuelve la lista de nombres cualificados de las tablas reconstruidas.
    """
    _rebuild_person_stats(conn)
    _rebuild_event_monthly(conn)
    _rebuild_contact_link_quality(conn)
    _rebuild_org_contacts(conn)
    return [f"derived.{name}" for name in DERIVED_TABLES]


def _rebuild_person_stats(conn) -> None:
    """Agregados de persons por contexto, pais y tag (sin note_path).

    Una fila por (dimension, valor) con el conteo de personas. Los tags se
    expanden en Python desde el JSON de la columna tags para no depender de
    funciones JSON del motor.
    """
    rows = conn.execute("SELECT contexto, pais, tags FROM persons").fetchall()
    counts: dict = {}
    for contexto, pais, tags_json in rows:
        counts[("contexto", contexto or "(sin)")] = (
            counts.get(("contexto", contexto or "(sin)"), 0) + 1
        )
        counts[("pais", pais or "(sin)")] = counts.get(("pais", pais or "(sin)"), 0) + 1
        try:
            tags = json.loads(tags_json) if tags_json else []
        except (TypeError, ValueError):
            tags = []
        if not isinstance(tags, list):
            tags = [tags]
        for tag in tags:
            key = ("tag", str(tag))
            counts[key] = counts.get(key, 0) + 1

    conn.execute("DROP TABLE IF EXISTS derived.person_stats")
    conn.execute(
        "CREATE TABLE derived.person_stats (dimension TEXT, valor TEXT, n BIGINT)"
    )
    payload = [[dim, val, n] for (dim, val), n in sorted(counts.items())]
    if payload:
        conn.executemany(
            "INSERT INTO derived.person_stats VALUES (?, ?, ?)", payload
        )


def _rebuild_event_monthly(conn) -> None:
    """Conteo de eventos por calendario y mes (sin note_path)."""
    conn.execute("DROP TABLE IF EXISTS derived.event_monthly")
    conn.execute(
        "CREATE TABLE derived.event_monthly AS "
        "SELECT calendar, substr(dtstart, 1, 7) AS month, COUNT(*) AS n "
        "FROM events WHERE dtstart IS NOT NULL "
        "GROUP BY calendar, substr(dtstart, 1, 7) ORDER BY calendar, month"
    )


def _rebuild_contact_link_quality(conn) -> None:
    """Calidad del enlace contacts -> persons: solo números, sin paths."""
    conn.execute("DROP TABLE IF EXISTS derived.contact_link_quality")
    conn.execute(
        "CREATE TABLE derived.contact_link_quality AS "
        "SELECT COUNT(*) AS total, "
        "COUNT(*) FILTER (WHERE note_path IS NOT NULL) AS linked, "
        "COUNT(*) FILTER (WHERE note_path IS NULL) AS unlinked "
        "FROM contacts"
    )


def _rebuild_org_contacts(conn) -> None:
    """Teléfonos de las personas de contacto de cada organización.

    Deriva del campo ``relaciones`` del frontmatter de cada persona (forma
    ``"[[org-slug]] — rol"``): por cada persona relacionada con una organización
    se emite una fila (org_slug, org_nombre, persona, rol, telefono) por cada
    teléfono que tenga la persona. Permite que, al buscar una empresa, aparezcan
    todos los teléfonos de sus contactos con el nombre de la persona y su rol.

    Solo datos computados (regla del módulo: sin note_path). El parseo del JSON
    de ``extra_fm`` y de los wikilinks se hace en Python para no depender de las
    funciones JSON del motor, igual que ``_rebuild_person_stats`` con los tags.
    """
    org_names = {
        slug: nombre
        for slug, nombre in conn.execute(
            "SELECT slug, nombre FROM organizations"
        ).fetchall()
    }
    persons = conn.execute(
        "SELECT nombre, telefono, telefonos, extra_fm FROM persons"
    ).fetchall()

    rows: list = []
    seen: set = set()  # (org_slug, persona, telefono) para no duplicar
    for nombre, telefono, telefonos_json, extra_fm in persons:
        if not extra_fm:
            continue
        try:
            fm = json.loads(extra_fm)
        except (TypeError, ValueError):
            continue
        relaciones = fm.get("relaciones") or []
        if not isinstance(relaciones, list):
            relaciones = [relaciones]

        try:
            tels = json.loads(telefonos_json) if telefonos_json else []
        except (TypeError, ValueError):
            tels = []
        if not isinstance(tels, list):
            tels = [tels]
        if not tels and telefono:
            tels = [telefono]
        tels = [str(t).strip() for t in tels if str(t).strip()]
        if not tels:
            continue

        for rel in relaciones:
            m = _RELACION_RE.match(str(rel))
            if not m:
                continue
            org_slug = m.group(1).strip()
            rol = m.group(2).strip() or "contacto"
            org_nombre = org_names.get(org_slug, org_slug)
            for tel in tels:
                key = (org_slug, nombre, tel)
                if key in seen:
                    continue
                seen.add(key)
                rows.append([org_slug, org_nombre, nombre, rol, tel])

    conn.execute("DROP TABLE IF EXISTS derived.org_contacts")
    conn.execute(
        "CREATE TABLE derived.org_contacts ("
        "org_slug TEXT, org_nombre TEXT, persona TEXT, rol TEXT, telefono TEXT)"
    )
    if rows:
        conn.executemany(
            "INSERT INTO derived.org_contacts VALUES (?, ?, ?, ?, ?)", rows
        )