"""Leitura do grafo: vizinhança de um NIF, ficha de entidade, cobertura.

A travessia é BFS por níveis em Python e não um CTE recursivo. O CTE fica
documentado no `registry_graph` como referência, mas para servir a UI o BFS ganha
em três coisas que importam aqui: dá a profundidade mínima exacta (um CTE
re-expande um nó alcançado por dois caminhos), permite tectos por nível em vez de
um tecto global, e é instrumentável — quando um grafo vem truncado é possível
dizer *onde* foi cortado.

Três tectos, e nenhum deles é decorativo:

* **profundidade** — 3 hops. Cada hop multiplica; ao quarto o grafo deixa de se
  ler.
* **grau do nó** — um gerente nomeado em 400 empresas, um banco como sócio, uma
  cooperativa com 300 associados. Atravessá-los liga tudo a tudo e o resultado
  deixa de significar nada. Não se escondem: aparecem como nó, marcados, mas não
  se expandem.
* **número de nós** — e quando se bate no tecto diz-se `truncated: true`, em vez
  de devolver metade do grafo com ar de estar completo.
"""
import logging
import time
from typing import Any

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.services.registry_act_parser import normalize_name

logger = logging.getLogger(__name__)

MAX_DEPTH = 3
MAX_NODES = 400
# Acima disto o nó entra no grafo mas não se atravessa.
HUB_DEGREE = 40

EDGE_TYPES = (
    "holds_quota", "manages", "audits", "secretary", "liquidator",
    "depositary", "spouse_of", "other",
)


def _norm_nif(raw: str | None) -> str | None:
    if not raw:
        return None
    digits = "".join(ch for ch in raw if ch.isdigit())
    return digits if len(digits) == 9 else None


async def _nodes_by_key(
    session: AsyncSession, keys: list[str], limit: int | None = None
) -> dict[str, dict[str, Any]]:
    """Nós indexados pela `entity_key`, não pelo NIF.

    Um titular estrangeiro ou não identificado não tem NIF — são 67 arestas
    actuais em 29 empresas. Indexados por NIF ficavam de fora do dicionário de
    nós, a aresta chegava ao frontend a apontar para o nada e era descartada em
    silêncio: a ficha listava o sócio e o diagrama não o desenhava, sem nada
    dizer que faltava ali alguém.

    A `entity_key` existe para todas as entidades — para as que têm NIF é
    derivada dele —, portanto passa a ser a chave da travessia.

    O `limit` traz a ordenação por relevância para dentro da consulta. A
    alternativa — trazer tudo e cortar em Python — obrigava a ler cinco mil
    linhas para ficar com quatrocentas, e mede-se: eram 400 ms a três hops.
    Monitorizadas primeiro, depois insolvências abertas, depois o grau.
    """
    if not keys:
        return {}
    ordem = (
        """
        ORDER BY (s.monitored_company_id IS NOT NULL) DESC,
                 COALESCE(s.insolvency_open, FALSE) DESC,
                 COALESCE(s.cire_ann_as_debtor, 0) DESC,
                 e.degree_current DESC
        LIMIT :lim
        """
        if limit
        else ""
    )
    rows = (
        await session.execute(
            text(
                """
                SELECT e.id::text, e.nif, e.entity_key, e.kind, e.name, e.degree_current,
                       e.natureza_juridica, e.concelho, e.distrito,
                       e.captable_act_id::text, e.captable_as_of,
                       e.captable_newer_act_date, e.captable_total,
                       e.captable_base, e.captable_unidentified,
                       e.captable_currency, e.captable_complete, e.acts_count,
                       e.history_fetched_at,
                       -- quantas quotas do acto é que a fonte não atribui: é a
                       -- diferença entre "falta 90 cêntimos" e "faltam quatro
                       -- quotas que valem 750.000 €"
                       ca.parse_quota_orphans AS captable_orphan_quotas,
                       ca.parse_capital AS captable_act_capital,
                       s.cire_ann_count, s.cire_ann_as_debtor, s.insolvency_open,
                       s.open_claim_deadline, s.process_count, s.risk_score,
                       s.monitored_company_id::text, s.monitoring_type
                  FROM registry_entities e
                  LEFT JOIN registry_acts ca ON ca.id = e.captable_act_id
                  LEFT JOIN entity_incident_summary s ON s.nif = e.nif
                 WHERE e.entity_key = ANY(:keys)
                """
                + ordem
            ),
            {"keys": keys, **({"lim": limit} if limit else {})},
        )
    ).mappings().all()
    saida: dict[str, dict[str, Any]] = {}
    for r in rows:
        no = dict(r)
        no["captable_superseded"], no["captable_superseded_reason"] = _captable_superseded(
            no.get("natureza_juridica"), no.get("captable_as_of")
        )
        saida[r["entity_key"]] = no
    return saida


# Formas jurídicas em que o capital **não** é dividido em quotas. Numa S.A. o
# capital são acções e o registo comercial não publica accionistas: não há, nem
# pode haver, uma cap table actual.
_SEM_QUOTAS = {"sociedade anonima"}


def _captable_superseded(
    natureza: str | None, captable_as_of: Any
) -> tuple[bool, str | None]:
    """A cap table de quotas guardada ainda descreve o presente?

    Uma sociedade que se transforma em anónima deixa de ter quotas. As arestas
    que temos continuam a ser factos verdadeiros do acto que as publicou — e é o
    que permite responder "onde é que este senhor foi sócio" —, mas apresentá-las
    como estrutura actual é afirmar uma coisa que deixou de existir.

    Foi assim que a JOIN THE MOMENT TRANSITÁRIOS mostrava um sócio de 2011 numa
    empresa que é S.A. desde 2022. São 1.020 empresas na mesma situação.
    """
    if captable_as_of is None or not natureza:
        return False, None
    if normalize_name(natureza) in _SEM_QUOTAS:
        return True, "sociedade anónima"
    return False, None


async def _edges_for(
    session: AsyncSession, entity_ids: list[str], edge_types: list[str]
) -> list[dict[str, Any]]:
    if not entity_ids:
        return []
    rows = (
        await session.execute(
            text(
                """
                SELECT r.id::text, r.edge_type, r.role_label, r.quota_amount,
                       r.currency, r.quota_share, r.valid_from, r.valid_to,
                       r.is_current, r.cease_cause, r.confidence,
                       h.nif AS holder_nif, h.name AS holder_name, h.kind AS holder_kind,
                       h.entity_key AS holder_key,
                       s.nif AS subject_nif, s.name AS subject_name, s.kind AS subject_kind,
                       s.entity_key AS subject_key,
                       a.id::text AS act_id, a.act_type, a.act_date
                  FROM registry_edges r
                  JOIN registry_entities h ON h.id = r.holder_id
                  JOIN registry_entities s ON s.id = r.subject_id
                  JOIN registry_acts a ON a.id = r.source_act_id
                 WHERE r.is_current
                   AND r.edge_type = ANY(:types)
                   AND (r.holder_id = ANY(CAST(:ids AS UUID[]))
                        OR r.subject_id = ANY(CAST(:ids AS UUID[])))
                """
            ),
            {"ids": entity_ids, "types": edge_types},
        )
    ).mappings().all()
    return [dict(r) for r in rows]


# A cobertura viaja em **todas** as respostas do grafo, e são cinco `count(*)`
# sobre tabelas que já vão em 934 mil actos: medido, custava 250 a 350 ms por
# chamada, enquanto a travessia em si custa 20. Um minuto de cache resolve-o sem
# mentir — estes números dizem que janela é que o índice cobre, e essa janela não
# muda de segundo a segundo.
_COVERAGE_TTL = 60.0
_coverage_cache: tuple[float, dict[str, Any]] | None = None


async def coverage(session: AsyncSession) -> dict[str, Any]:
    """O que o índice já cobre.

    Vai a par de qualquer resposta que mostre as participações de uma pessoa: essa
    pergunta só é respondível pelo índice nacional, e a resposta vale exactamente
    o que valer a janela recolhida. Mostrar participações sem dizer o intervalo
    coberto convida a ler ausência de dados como ausência de participações.
    """
    global _coverage_cache
    agora = time.monotonic()
    if _coverage_cache and agora - _coverage_cache[0] < _COVERAGE_TTL:
        return _coverage_cache[1]
    row = (
        await session.execute(
            text(
                """
                SELECT
                  (SELECT min(day) FROM registry_sweep_days WHERE status = 'ok') AS varrido_de,
                  (SELECT max(day) FROM registry_sweep_days WHERE status = 'ok') AS varrido_ate,
                  (SELECT count(*) FROM registry_sweep_days WHERE status = 'ok') AS dias_ok,
                  (SELECT count(*) FROM registry_sweep_days
                    WHERE status IN ('pending','error','partial')) AS dias_em_falta,
                  (SELECT count(*) FROM registry_acts) AS actos,
                  (SELECT count(*) FROM registry_acts WHERE body IS NOT NULL) AS actos_lidos,
                  (SELECT count(*) FROM registry_entities) AS entidades,
                  (SELECT count(*) FROM registry_entities
                    WHERE history_fetched_at IS NOT NULL) AS entidades_com_historico,
                  (SELECT count(*) FROM registry_edges WHERE is_current) AS arestas
                """
            )
        )
    ).mappings().first()
    out = dict(row) if row else {}
    _coverage_cache = (agora, out)
    return out


async def neighbourhood(
    session: AsyncSession,
    nif: str,
    *,
    depth: int = 2,
    edge_types: list[str] | None = None,
    include_hubs: bool = False,
) -> dict[str, Any]:
    """Grafo em volta de um NIF, com nós, arestas e o que ficou de fora."""
    norm = _norm_nif(nif)
    if not norm:
        return {"nodes": [], "edges": [], "meta": {"error": "NIF inválido"}}
    depth = max(0, min(depth, MAX_DEPTH))
    types = [t for t in (edge_types or list(EDGE_TYPES)) if t in EDGE_TYPES]

    # A raiz entra por NIF — é o que a URL traz — mas a partir daqui tudo anda
    # pela `entity_key`, para os nós sem NIF não caírem da travessia.
    root = (
        await session.execute(
            text("SELECT entity_key FROM registry_entities WHERE nif = :nif"),
            {"nif": norm},
        )
    ).scalar()
    known = await _nodes_by_key(session, [root]) if root else {}
    if not root or root not in known:
        return {
            "nodes": [], "edges": [],
            "meta": {"root": norm, "found": False, "coverage": await coverage(session)},
        }

    nodes: dict[str, dict[str, Any]] = {}
    edges: dict[str, dict[str, Any]] = {}
    hubs: list[dict[str, Any]] = []
    truncated = False

    node = dict(known[root])
    node["depth"] = 0
    nodes[root] = node
    frontier = [root]

    for level in range(depth):
        ids = [nodes[n]["id"] for n in frontier]
        if not ids:
            break
        na_fronteira = set(frontier)

        found = await _edges_for(session, ids, types)

        # O grau que decide se um nó é hub tem de ser contado **nos tipos que
        # estão a ser vistos**, e não no total. A PWC audita 430 empresas: com o
        # `degree_current` (506) era hub mesmo numa vista de quotas e gerência,
        # onde tem 54 ligações e é perfeitamente desenhável. Conta-se sobre as
        # arestas já lidas, portanto não custa consulta nenhuma.
        grau: dict[str, int] = {}
        por_no: dict[str, list[dict[str, Any]]] = {}
        for edge in found:
            edges[edge["id"]] = edge
            for side in ("holder", "subject"):
                key = edge[f"{side}_key"]
                if key in na_fronteira:
                    grau[key] = grau.get(key, 0) + 1
                    por_no.setdefault(key, []).append(edge)

        next_keys: set[str] = set()
        for key, ligacoes in por_no.items():
            hub = grau.get(key, 0) > HUB_DEGREE and not include_hubs and level > 0
            if hub:
                hubs.append({
                    "nif": nodes[key]["nif"], "name": nodes[key]["name"],
                    "degree": grau[key],
                })
                continue
            for edge in ligacoes:
                for side in ("holder", "subject"):
                    other = edge[f"{side}_key"]
                    if other and other not in nodes:
                        next_keys.add(other)

        if not next_keys:
            break

        # O corte é por relevância e não por ordem de chave: antes ficava de
        # fora um concorrente com insolvência a favor de uma entidade anónima só
        # porque a chave dela vinha antes no alfabeto. A ordenação vai no SQL
        # para não se lerem linhas que vão ser deitadas fora.
        espaco = max(0, MAX_NODES - len(nodes))
        fetched = await _nodes_by_key(session, sorted(next_keys), limit=espaco)
        if len(next_keys) > espaco:
            truncated = True

        frontier = []
        for data in fetched.values():
            data = dict(data)
            data["depth"] = level + 1
            nodes[data["entity_key"]] = data
            frontier.append(data["entity_key"])
        if truncated or not frontier:
            break

    # Arestas cujas duas pontas estão no grafo. Uma aresta pendurada num nó que
    # ficou de fora do tecto desenharia uma ligação para o nada.
    keys = set(nodes)
    kept = [
        e for e in edges.values()
        if e["holder_key"] in keys and e["subject_key"] in keys
    ]

    return {
        "nodes": list(nodes.values()),
        "edges": kept,
        "meta": {
            "root": norm,
            "found": True,
            "depth": depth,
            "node_count": len(nodes),
            "edge_count": len(kept),
            "truncated": truncated,
            "hubs_skipped": hubs,
            "hub_degree": HUB_DEGREE,
            "coverage": await coverage(session),
        },
    }


async def entity_detail(session: AsyncSession, nif: str) -> dict[str, Any] | None:
    """Ficha completa de uma entidade, monitorizada ou não.

    É este endpoint que faz o grafo valer a pena: um nó descoberto a dois hops de
    distância tem ficha, cap table e incidentes sem ninguém ter de o registar
    primeiro.
    """
    norm = _norm_nif(nif)
    if not norm:
        return None
    # A ficha entra sempre por NIF: uma entidade sem NIF não tem página própria,
    # aparece na de quem a nomeia.
    key = (
        await session.execute(
            text("SELECT entity_key FROM registry_entities WHERE nif = :nif"),
            {"nif": norm},
        )
    ).scalar()
    base = await _nodes_by_key(session, [key]) if key else {}
    if not key or key not in base:
        return None
    entity = dict(base[key])

    # Uma linha por **titular**, não por quota. O mesmo sócio pode deter várias
    # quotas — a RAMAZON detém cinco da CENTRALMED —, e listar cada uma como se
    # fosse um sócio dá cinco linhas com 40%, 27,6%, 12,4%, 15% e 5% quando a
    # verdade é um sócio único com 100%.
    captable = (
        await session.execute(
            text(
                """
                SELECT h.nif, h.name, h.kind,
                       sum(r.quota_amount) AS quota_amount,
                       min(r.currency) AS currency,
                       sum(r.quota_share) AS quota_share,
                       count(*) AS quotas,
                       min(r.valid_from) AS valid_from,
                       max(a.act_type) AS act_type, max(a.act_date) AS act_date,
                       max(a.id::text) AS act_id, min(r.confidence) AS confidence
                  FROM registry_edges r
                  JOIN registry_entities h ON h.id = r.holder_id
                  JOIN registry_acts a ON a.id = r.source_act_id
                 WHERE r.subject_id = CAST(:id AS UUID)
                   AND r.edge_type = 'holds_quota' AND r.is_current
                 GROUP BY h.id, h.nif, h.name, h.kind
                 ORDER BY sum(r.quota_amount) DESC NULLS LAST
                """
            ),
            {"id": entity["id"]},
        )
    ).mappings().all()

    # O que mudou entre a estrutura anterior e a actual. `is_current = false`
    # numa quota quer dizer que **aquele acto foi substituído**, não que a pessoa
    # saiu: na SEGUNOR o José Ribeiro tem uma linha de 125.000 € de 2019 e outra
    # de 155.000 € de 2026 — aumentou a quota. Só quem não tem nenhuma aresta
    # actual é que saiu de facto.
    changes: dict[str, list[dict[str, Any]]] = {"entraram": [], "sairam": [], "alteraram": []}
    # Só há "o que mudou" se houver um antes. Sem um acto anterior que sirva de
    # estrutura, a comparação daria toda a gente a "entrar" — que é o que a
    # GÁLIA mostrava: cinco sócios apresentados como entradas por não existir
    # nenhuma fotografia anterior com que os comparar.
    #
    # O acto anterior tem de valer como estrutura, senão compara-se contra uma
    # fotografia que o próprio sistema reprovou e inventam-se saídas. Mesmos dois
    # degraus do `registry_graph._captable_act`: elegível primeiro, qualquer um
    # com quotas depois.
    previous_act = None
    if entity.get("captable_act_id"):
        previous_act = (
            await session.execute(
                text(
                    """
                    SELECT a.id::text
                      FROM registry_acts a
                     WHERE a.nipc = :nipc
                       AND COALESCE(a.parse_quotas, 0) > 0
                       AND a.id <> CAST(:act AS UUID)
                       AND a.act_date < CAST(:asof AS DATE)
                     ORDER BY COALESCE(a.parse_captable_eligible, FALSE) DESC,
                              a.act_date DESC
                     LIMIT 1
                    """
                ),
                {"nipc": norm, "act": entity["captable_act_id"],
                 "asof": entity["captable_as_of"]},
            )
        ).scalar()
    if previous_act:
        previous = (
            await session.execute(
                text(
                    """
                    WITH anterior AS (
                        SELECT r.holder_id,
                               sum(r.quota_amount) AS quota_amount,
                               min(r.currency) AS currency,
                               max(a.act_date) AS act_date
                          FROM registry_edges r
                          JOIN registry_acts a ON a.id = r.source_act_id
                         WHERE r.subject_id = CAST(:id AS UUID)
                           AND r.edge_type = 'holds_quota'
                           AND r.source_act_id = CAST(:prev AS UUID)
                         GROUP BY r.holder_id
                    ),
                    atual AS (
                        SELECT r.holder_id,
                               sum(r.quota_amount) AS quota_amount,
                               min(r.currency) AS currency
                          FROM registry_edges r
                         WHERE r.subject_id = CAST(:id AS UUID)
                           AND r.edge_type = 'holds_quota' AND r.is_current
                         GROUP BY r.holder_id
                    )
                    SELECT h.nif, h.name, h.kind,
                           p.quota_amount AS antes, c.quota_amount AS agora,
                           COALESCE(c.currency, p.currency) AS currency,
                           p.currency AS moeda_anterior,
                           p.act_date AS data_anterior
                      FROM anterior p
                      FULL OUTER JOIN atual c ON c.holder_id = p.holder_id
                      JOIN registry_entities h ON h.id = COALESCE(c.holder_id, p.holder_id)
                     ORDER BY COALESCE(c.quota_amount, p.quota_amount) DESC NULLS LAST
                    """
                ),
                {"id": entity["id"], "prev": previous_act},
            )
        ).mappings().all()
        for row in previous:
            d = dict(row)
            # Escudos para euros não é uma redução de 99,5%: é a mesma quota
            # noutra moeda. Vai para "alterou" com a mudança assinalada, e quem
            # apresenta decide se compara valores.
            d["moeda_mudou"] = bool(
                d["antes"] is not None
                and d["agora"] is not None
                and d["moeda_anterior"]
                and d["currency"]
                and d["moeda_anterior"] != d["currency"]
            )
            if d["antes"] is None and d["agora"] is not None:
                changes["entraram"].append(d)
            elif d["agora"] is None and d["antes"] is not None:
                changes["sairam"].append(d)
            elif d["antes"] != d["agora"] or d["moeda_mudou"]:
                changes["alteraram"].append(d)

    officers = (
        await session.execute(
            text(
                """
                SELECT h.nif, h.name, h.kind, r.edge_type, r.role_label,
                       r.valid_from, r.valid_to, r.is_current, r.cease_cause,
                       a.act_type, a.act_date
                  FROM registry_edges r
                  JOIN registry_entities h ON h.id = r.holder_id
                  JOIN registry_acts a ON a.id = r.source_act_id
                 WHERE r.subject_id = CAST(:id AS UUID)
                   AND r.edge_type NOT IN ('holds_quota', 'spouse_of')
                 ORDER BY r.is_current DESC, r.valid_from DESC NULLS LAST
                 LIMIT 60
                """
            ),
            {"id": entity["id"]},
        )
    ).mappings().all()

    # Participações **desta** entidade noutras: o outro sentido da aresta, que é
    # o que responde a "e em que mais é que ele tem quotas".
    holdings = (
        await session.execute(
            text(
                """
                SELECT s.nif, s.name, s.kind, r.edge_type, r.role_label,
                       sum(r.quota_amount) AS quota_amount,
                       min(r.currency) AS currency,
                       sum(r.quota_share) AS quota_share,
                       count(*) FILTER (WHERE r.edge_type = 'holds_quota') AS quotas,
                       min(r.valid_from) AS valid_from,
                       max(a.act_type) AS act_type, max(a.act_date) AS act_date,
                       bool_or(i.insolvency_open) AS insolvency_open,
                       max(i.cire_ann_as_debtor) AS cire_ann_as_debtor,
                       min(i.monitoring_type) AS monitoring_type
                  FROM registry_edges r
                  JOIN registry_entities s ON s.id = r.subject_id
                  JOIN registry_acts a ON a.id = r.source_act_id
                  LEFT JOIN entity_incident_summary i ON i.nif = s.nif
                 WHERE r.holder_id = CAST(:id AS UUID)
                   AND r.is_current AND r.edge_type <> 'spouse_of'
                 GROUP BY s.id, s.nif, s.name, s.kind, r.edge_type, r.role_label
                 ORDER BY r.edge_type, sum(r.quota_amount) DESC NULLS LAST
                 LIMIT 100
                """
            ),
            {"id": entity["id"]},
        )
    ).mappings().all()

    spouses = (
        await session.execute(
            text(
                """
                -- Uma linha por cônjuge, não uma por acto onde ele aparece. A
                -- aresta repete-se de propósito — cada uma é um facto de um acto
                -- diferente e não se apaga —, mas a mesma pessoa listada seis
                -- vezes lê-se como seis cônjuges.
                SELECT * FROM (
                    SELECT DISTINCT ON (h.id)
                           h.nif, h.name, h.entity_key, r.role_label AS regime,
                           a.act_date,
                           (SELECT count(*) FROM registry_identity_links l
                             WHERE l.unknown_id = h.id AND l.confirmed_at IS NULL
                               AND l.rejected_at IS NULL) AS candidatos
                      FROM registry_edges r
                      JOIN registry_entities h ON h.id = r.holder_id
                      JOIN registry_acts a ON a.id = r.source_act_id
                     WHERE r.subject_id = CAST(:id AS UUID)
                       AND r.edge_type = 'spouse_of'
                     ORDER BY h.id, a.act_date DESC NULLS LAST
                ) t
                 ORDER BY act_date DESC NULLS LAST
                 LIMIT 20
                """
            ),
            {"id": entity["id"]},
        )
    ).mappings().all()

    acts = (
        await session.execute(
            text(
                """
                SELECT id::text, act_type, act_class, act_date, apresentacao,
                       parse_quotas, parse_people, parse_quota_orphans,
                       parse_capital, parse_captable_complete, parse_error,
                       (body IS NOT NULL) AS lido
                  FROM registry_acts WHERE nipc = :nif
                 ORDER BY act_date DESC NULLS LAST
                 LIMIT 80
                """
            ),
            {"nif": norm},
        )
    ).mappings().all()

    return {
        "entity": entity,
        "captable": [dict(r) for r in captable],
        "captable_changes": changes,
        "officers": [dict(r) for r in officers],
        "holdings": [dict(r) for r in holdings],
        "spouses": [dict(r) for r in spouses],
        "acts": [dict(r) for r in acts],
        "coverage": await coverage(session),
    }


async def search_entities(
    session: AsyncSession, query: str, limit: int = 30
) -> list[dict[str, Any]]:
    """Pesquisa por NIF exacto ou por nome (trigram)."""
    norm = _norm_nif(query)
    if norm:
        rows = (
            await session.execute(
                text(
                    """
                    SELECT e.nif, e.name, e.kind, e.concelho, e.degree_current,
                           i.monitoring_type, i.insolvency_open
                      FROM registry_entities e
                      LEFT JOIN entity_incident_summary i ON i.nif = e.nif
                     WHERE e.nif = :nif
                    """
                ),
                {"nif": norm},
            )
        ).mappings().all()
        if rows:
            return [dict(r) for r in rows]

    # O trigram sozinho não chega para nomes compridos, que é a norma nas firmas.
    # Medido: "securitas" contra "securitas servicos e tecnologia de seguranca sa"
    # dá similaridade **0,233**, abaixo do limiar de 0,3 do `%` — ou seja, procurar
    # "Securitas" não encontrava a Securitas, só as empresas de nome curto. O
    # `ILIKE` apanha o caso de se escrever uma parte distintiva do nome, que é
    # como as pessoas procuram.
    term = query.lower().strip()
    rows = (
        await session.execute(
            text(
                """
                SELECT e.nif, e.name, e.kind, e.concelho, e.degree_current,
                       i.monitoring_type, i.insolvency_open,
                       GREATEST(similarity(e.name_norm, :q),
                                CASE WHEN e.name_norm ILIKE :like THEN 0.6 ELSE 0 END
                       ) AS score
                  FROM registry_entities e
                  LEFT JOIN entity_incident_summary i ON i.nif = e.nif
                 WHERE e.name_norm ILIKE :like OR e.name_norm % :q
                 ORDER BY score DESC, e.degree_current DESC
                 LIMIT :lim
                """
            ),
            {"q": term, "like": f"%{term}%", "lim": limit},
        )
    ).mappings().all()
    return [dict(r) for r in rows]


async def entry_points(session: AsyncSession, limit: int = 24) -> list[dict[str, Any]]:
    """Nós por onde começar, para a página da rede não abrir vazia.

    Uma página que só mostra uma caixa de pesquisa parece avariada, sobretudo
    quando não se sabe o que lá há dentro. Estas são as entidades monitorizadas
    com mais ligações conhecidas — que é por onde alguém quereria começar de
    qualquer forma.
    """
    rows = (
        await session.execute(
            text(
                """
                SELECT e.nif, e.name, e.kind, e.concelho, e.degree_current,
                       i.monitoring_type, i.insolvency_open,
                       e.captable_as_of
                  FROM registry_entities e
                  JOIN entity_incident_summary i ON i.nif = e.nif
                 WHERE i.monitoring_type IS NOT NULL
                   AND e.degree_current > 0
                 ORDER BY e.degree_current DESC, e.name
                 LIMIT :lim
                """
            ),
            {"lim": limit},
        )
    ).mappings().all()
    return [dict(r) for r in rows]


async def refresh_incident_summary(session: AsyncSession) -> int:
    """Incidentes por NIF, para pintar os nós do grafo.

    O `cire_announcement_parties` é **nacional** — 240 296 partes, 55 321 NIFs
    distintos — portanto isto funciona para pessoas e entidades que não
    monitorizamos, que é precisamente o que faltava. Já `processes` é só das
    monitorizadas, e por isso entra como zero em quem não é.
    """
    result = await session.execute(
        text(
            """
            WITH cire AS (
                SELECT p.nif,
                       count(*) AS total,
                       count(*) FILTER (
                         WHERE lower(COALESCE(p.role,'')) IN ('insolvente','devedor','requerido')
                       ) AS as_debtor,
                       max(a.pub_date) AS last_date
                  FROM cire_announcement_parties p
                  JOIN cire_announcements a ON a.id = p.announcement_id
                 WHERE p.nif IS NOT NULL
                 GROUP BY p.nif
            ),
            hits AS (
                SELECT h.matched_nif AS nif,
                       min(h.claim_deadline) FILTER (WHERE h.claim_deadline >= CURRENT_DATE)
                         AS open_deadline
                  FROM insolvency_hits h
                 GROUP BY h.matched_nif
            ),
            procs AS (
                SELECT c.nif, count(*) AS total, max(p.date_filed) AS last_date
                  FROM processes p JOIN companies c ON c.id = p.company_id
                 GROUP BY c.nif
            ),
            contratos AS (
                SELECT c.nif, count(*) AS total
                  FROM public_contracts pc JOIN companies c ON c.id = pc.company_id
                 GROUP BY c.nif
            ),
            todos AS (
                SELECT nif FROM cire
                UNION SELECT nif FROM hits
                UNION SELECT nif FROM procs
                UNION SELECT nif FROM contratos
                UNION SELECT nif FROM companies
                UNION SELECT nif FROM registry_entities WHERE nif IS NOT NULL
            )
            INSERT INTO entity_incident_summary (
                nif, cire_ann_count, cire_ann_as_debtor, cire_ann_last_date,
                insolvency_open, open_claim_deadline, process_count, process_last_date,
                contracts_count, risk_score, monitored_company_id, monitoring_type,
                computed_at
            )
            SELECT t.nif,
                   COALESCE(ci.total, 0), COALESCE(ci.as_debtor, 0), ci.last_date,
                   (h.open_deadline IS NOT NULL), h.open_deadline,
                   COALESCE(pr.total, 0), pr.last_date,
                   COALESCE(co.total, 0),
                   co2.risk_score, co2.id, co2.monitoring_type,
                   now()
              FROM todos t
              LEFT JOIN cire ci ON ci.nif = t.nif
              LEFT JOIN hits h ON h.nif = t.nif
              LEFT JOIN procs pr ON pr.nif = t.nif
              LEFT JOIN contratos co ON co.nif = t.nif
              LEFT JOIN companies co2 ON co2.nif = t.nif
             WHERE t.nif IS NOT NULL AND length(trim(t.nif)) = 9
            ON CONFLICT (nif) DO UPDATE SET
                cire_ann_count = EXCLUDED.cire_ann_count,
                cire_ann_as_debtor = EXCLUDED.cire_ann_as_debtor,
                cire_ann_last_date = EXCLUDED.cire_ann_last_date,
                insolvency_open = EXCLUDED.insolvency_open,
                open_claim_deadline = EXCLUDED.open_claim_deadline,
                process_count = EXCLUDED.process_count,
                process_last_date = EXCLUDED.process_last_date,
                contracts_count = EXCLUDED.contracts_count,
                risk_score = EXCLUDED.risk_score,
                monitored_company_id = EXCLUDED.monitored_company_id,
                monitoring_type = EXCLUDED.monitoring_type,
                computed_at = now()
            """
        )
    )
    return result.rowcount or 0