"""Estado actual do grafo: quem detém o quê hoje, e quem já saiu. Duas regras, uma por família de facto, e a diferença entre elas é a única coisa que aqui importa entender. **Quotas — a fotografia mais recente vale por inteiro.** Os actos de cap table (`SÓCIOS E QUOTAS:`) publicam a estrutura *completa* da sociedade, não o delta. Portanto a cap table actual de uma empresa é exactamente o conjunto de arestas do acto mais recente em que o parser **encontrou** titulares. Não se reconstrói por deltas, e não se faz merge entre actos: é o mesmo princípio já defendido no `mj_company_enrichment.merge_company_fields` para a identidade e a morada, e a propriedade merece-o mais ainda. A elegibilidade é `parse_quotas > 0`, **nunca o título do acto**. Umas "alterações ao contrato" que só mudaram a sede não têm titulares; tratá-las como fotografia vazia apagaria os sócios da empresa em silêncio, e ficaria com aspecto normal. **Cargos — base mais última palavra por par.** A constituição lista o órgão completo e serve de base; cada acto posterior sobrepõe-se apenas ao par `(pessoa, empresa, tipo)` que menciona. Uma designação abre a aresta, uma cessação fecha-a. Um acto em falta corrompe só os pares que esse acto mencionava, nunca o quadro todo — que é a razão para não replicar aqui a lógica de deltas. Nada nesta camada apaga arestas. O histórico é o produto: só o reparse apaga, por `source_act_id`, e reinsere. """ import asyncio import logging from typing import Any from uuid import UUID from sqlalchemy import text from sqlalchemy.exc import DBAPIError from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) # `40P01` — o Postgres escolheu esta transacção como vítima de um impasse. _IMPASSE = "40P01" def _e_impasse(erro: BaseException) -> bool: return getattr(getattr(erro, "orig", None), "sqlstate", None) == _IMPASSE # Um sócio que passe a deter menos de metade do valor total, ou uma cap table que # fique sem ninguém, é sinal de parser e não de negócio. Não bloqueia — uma # redução de capital de 80 % é legítima — mas fica registado para a vigia diária. WIPEOUT_RATIO = 0.5 async def _captable_act(session: AsyncSession, nipc: str) -> dict[str, Any] | None: """O acto que define a estrutura actual, em dois degraus. Primeiro o mais recente **elegível**: com titulares, todos com valor e nenhuma quota órfã. Uma quota que a fonte lista sem nomear o titular desqualifica o acto para este efeito — foi o caso do `510036899`, onde usar o acto mais recente fazia a empresa perder o maior sócio (213.684,21 em três titulares passava a 10.685,21 em dois) sem nada parecer errado. Se nenhum for elegível, usa-se o mais recente com quotas e a estrutura fica marcada como incompleta. Mais vale uma estrutura declaradamente parcial do que nenhuma. """ for eligible_only in (True, False): clause = "AND parse_captable_eligible" if eligible_only else "" row = ( await session.execute( text( f""" SELECT id::text, COALESCE(parse_effective_date, act_date) AS act_date, parse_capital, parse_captable_complete, parse_quota_orphans, parse_quota_orphan_total, COALESCE(parse_captable_eligible, FALSE) AS eligible FROM registry_acts WHERE nipc = :nipc AND COALESCE(parse_quotas, 0) > 0 {clause} ORDER BY COALESCE(parse_effective_date, act_date) DESC NULLS LAST, first_seen_at DESC LIMIT 1 """ ), {"nipc": nipc}, ) ).mappings().first() if row: return dict(row) return None async def _newer_captable_act_date(session: AsyncSession, nipc: str, used: Any) -> Any: """Data do acto com quotas mais recente que o que foi usado, se houver. É o que permite dizer "estrutura verificada a 2020-05-12, com alterações posteriores em 2023-07-31" em vez de mostrar uma data sozinha e deixar quem lê a assumir que é o presente. """ if used is None: return None row = ( await session.execute( text( """ SELECT max(COALESCE(parse_effective_date, act_date)) FROM registry_acts WHERE nipc = :nipc AND COALESCE(parse_quotas, 0) > 0 AND COALESCE(parse_effective_date, act_date) > CAST(:used AS DATE) """ ), {"nipc": nipc, "used": used}, ) ).first() return row[0] if row else None def _share_base(act: dict[str, Any], identified: float | None) -> tuple[float | None, float | None]: """Sobre o que se calculam as percentagens, e quanto fica por explicar. Por ordem de preferência: 1. **o capital declarado no acto**, quando é maior ou igual à soma das quotas com titular. É o denominador certo: o que falta é capital de alguém que a publicação não nomeia, não capital que não existe; 2. **a soma das quotas com titular mais as órfãs**, quando o capital não aparece no acto mas as quotas anónimas têm valor; 3. **só o que foi identificado**, quando não há mais nada. A guarda do primeiro caso importa: em actos de aumento, o `parse_capital` é por vezes o *montante do aumento* e não o capital final. Aí é menor que a soma das quotas, e usá-lo daria percentagens acima de 100. """ if identified is None: return None, None capital = act.get("parse_capital") orphan_total = act.get("parse_quota_orphan_total") if capital is not None and float(capital) + 0.02 >= identified: base = float(capital) elif orphan_total is not None: base = identified + float(orphan_total) else: return identified, None falta = round(base - identified, 2) return base, (falta if falta > 0.02 else None) # Quem está em funções, por par pessoa-empresa. Uma definição só, usada pelo # caminho por entidade (`recompute_current`) e pelo recálculo global # (`recompute_roles`) — duas cópias divergiriam na primeira alteração. # # O `role_label` **não** entra na partição, e é a correcção central. É texto livre # da fonte: a mesma pessoa aparece como `presidente` num acto e `Presidente` no # seguinte, e cada grafia ganhava vida própria com o seu "mais recente". A renúncia # fechava uma e deixava a outra em funções para sempre. # # `dense_rank` e não `row_number`: há actos que atribuem dois cargos ao mesmo par # ("Administrador não executivo" e "Presidente" na mesma publicação). Com o # `source_act_id` no fim da ordenação, todas as linhas do acto vencedor ficam com # `rn = 1` em vez de uma delas perder. O desempate final entre actos distintos com # a mesma data é arbitrário mas estável — e nesse caso a pessoa fica em funções de # qualquer maneira, só muda qual o cargo mostrado. _ROLES_SCOPE_ONE = "AND subject_id = CAST(:sid AS UUID)" _ROLES_CURRENT_SQL = """ WITH alcance AS ( SELECT id, subject_id, holder_id, edge_type, role_label, valid_from, valid_to, source_act_id, snapshot_seq FROM registry_edges WHERE edge_type NOT IN ('holds_quota', 'spouse_of') {scope} ), -- Quando a fonte não repete o cargo não há texto para classificar e a aresta -- cai em `other`: é família **desconhecida**, não uma família diferente. Vale -- nos dois sentidos — uma cessação `other` fecha qualquer designação anterior -- do mesmo par, e uma designação `other` é fechada por qualquer cessação -- posterior. Na IGPS PROTEK são as duas pontas do mesmo caso: a renúncia do -- Luís veio sem cargo, e a designação do Carlos também. fecho AS ( SELECT subject_id, holder_id, max(valid_to) FILTER (WHERE edge_type = 'other') AS ate_other, max(valid_to) AS ate_qualquer FROM alcance WHERE valid_to IS NOT NULL GROUP BY subject_id, holder_id ), ranked AS ( SELECT id, subject_id, holder_id, edge_type, valid_from, valid_to, dense_rank() OVER ( PARTITION BY subject_id, holder_id, edge_type ORDER BY valid_from DESC NULLS LAST, (valid_to IS NOT NULL) DESC, source_act_id ) AS rn, -- Dentro do acto vencedor, uma linha por cargo distinto. Sem isto -- as repetições exactas — o mesmo acto que nomeia a mesma pessoa -- duas vezes, com `snapshot_seq` 0 e 1 — ficavam ambas em funções e -- a ficha da empresa mostrava a pessoa a dobrar. row_number() OVER ( PARTITION BY subject_id, holder_id, edge_type, COALESCE(role_label, '') ORDER BY valid_from DESC NULLS LAST, (valid_to IS NOT NULL) DESC, source_act_id, snapshot_seq ) AS rn_cargo FROM alcance ), veredicto AS ( SELECT r.id, (r.rn = 1 AND r.rn_cargo = 1 AND r.valid_to IS NULL AND (f.ate_other IS NULL OR r.valid_from IS NULL OR f.ate_other <= r.valid_from) AND (r.edge_type <> 'other' OR f.ate_qualquer IS NULL OR r.valid_from IS NULL OR f.ate_qualquer <= r.valid_from)) AS actual FROM ranked r LEFT JOIN fecho f ON f.subject_id = r.subject_id AND f.holder_id = r.holder_id ) UPDATE registry_edges e SET is_current = v.actual FROM veredicto v WHERE e.id = v.id AND e.is_current IS DISTINCT FROM v.actual """ async def recompute_roles(session: AsyncSession) -> int: """O mesmo cálculo de `recompute_current`, mas para toda a base de uma vez. São 276 mil empresas com arestas de cargo: um ciclo a chamar `recompute_current` uma vez por empresa não acabaria em tempo útil. Escreve só as linhas que mudam, portanto correr duas vezes seguidas não faz nada à segunda. """ result = await session.execute(text(_ROLES_CURRENT_SQL.format(scope=""))) return result.rowcount or 0 async def recompute_current(session: AsyncSession, subject_id: str, nipc: str) -> dict[str, Any]: """Recalcula `is_current` de uma entidade. Idempotente.""" out: dict[str, Any] = {"captable_act_id": None, "quota_edges": 0, "wipeout": False} before = ( await session.execute( text( """ SELECT COALESCE(sum(quota_amount), 0) AS total, count(*) AS n FROM registry_edges WHERE subject_id = CAST(:sid AS UUID) AND edge_type = 'holds_quota' AND is_current """ ), {"sid": subject_id}, ) ).mappings().first() act = await _captable_act(session, nipc) if act: await session.execute( text( """ UPDATE registry_edges SET is_current = (source_act_id = CAST(:aid AS UUID)), valid_to = CASE WHEN source_act_id = CAST(:aid AS UUID) THEN NULL ELSE COALESCE(valid_to, CAST(:adate AS DATE)) END WHERE subject_id = CAST(:sid AS UUID) AND edge_type = 'holds_quota' """ ), {"sid": subject_id, "aid": act["id"], "adate": act["act_date"]}, ) after = ( await session.execute( text( """ SELECT COALESCE(sum(quota_amount), 0) AS total, count(*) AS n, count(DISTINCT currency) AS moedas, min(currency) AS currency FROM registry_edges WHERE subject_id = CAST(:sid AS UUID) AND edge_type = 'holds_quota' AND is_current """ ), {"sid": subject_id}, ) ).mappings().first() # Vigia de apagão. Não impede a escrita: registar e deixar visível é mais # útil do que recusar, porque a causa legítima existe (amortização de # quotas, redução de capital) e a ilegítima é sempre um parser a falhar. if before and before["n"] and after: lost_all = after["n"] == 0 lost_half = ( float(before["total"] or 0) > 0 and float(after["total"] or 0) < float(before["total"]) * WIPEOUT_RATIO ) if lost_all or lost_half: out["wipeout"] = True logger.error( "cap table de %s encolheu: %s holders/%s -> %s holders/%s (acto %s)", nipc, before["n"], before["total"], after["n"], after["total"], act["id"], ) total = float(after["total"]) if after and after["n"] else None currency = after["currency"] if after and after["moedas"] == 1 else None newer = await _newer_captable_act_date(session, nipc, act["act_date"]) base, unidentified = _share_base(act, total) await session.execute( text( """ UPDATE registry_entities SET captable_act_id = CAST(:aid AS UUID), captable_as_of = CAST(:adate AS DATE), captable_newer_act_date = CAST(:newer AS DATE), captable_total = :total, captable_base = :base, captable_unidentified = :unidentified, captable_currency = :currency, captable_complete = COALESCE(:complete, FALSE), updated_at = now() WHERE id = CAST(:sid AS UUID) """ ), { "sid": subject_id, "aid": act["id"], "adate": act["act_date"], "newer": newer, "total": total, "base": base, "unidentified": unidentified, "currency": currency, "complete": act["parse_captable_complete"], }, ) out["newer_act_date"] = newer out["unidentified"] = unidentified # A percentagem é sempre dentro do mesmo acto — calcular contra o capital # de outro daria percentagens que não somam 100 — mas o denominador é o # capital desse acto e não a soma das quotas que têm titular. Ver # `_share_base`: dividir pelo que foi identificado é fingir que as quotas # anónimas não existem, e foi o que pôs a POLYEDRA com um sócio de 100 € # a deter 100% de uma sociedade de 4,7 milhões. await session.execute( text( """ UPDATE registry_edges e SET quota_share = ROUND( (e.quota_amount / NULLIF(CAST(:base AS NUMERIC), 0))::numeric, 4) WHERE e.source_act_id = CAST(:aid AS UUID) AND e.edge_type = 'holds_quota' AND e.quota_amount IS NOT NULL """ ), {"aid": act["id"], "base": base}, ) out["captable_act_id"] = act["id"] out["quota_edges"] = after["n"] if after else 0 # Cargos desta empresa. `spouse_of` fica de fora — é uma relação entre pessoas, # não um cargo que cesse. A regra está em `_ROLES_CURRENT_SQL`. await session.execute(text(_ROLES_CURRENT_SQL.format(scope=_ROLES_SCOPE_ONE)), {"sid": subject_id}) return out async def refresh_degrees(session: AsyncSession, entity_ids: list[str] | None = None) -> int: """Grau actual de cada nó, para a guarda anti-hub da travessia. Uma agregação só, e não uma subconsulta por entidade. A versão anterior percorria as arestas uma vez **por cada nó**: com 4 mil entidades não se notava, com as 42 mil de hoje e o índice nacional ainda a criá-las passava a doer, e cresce ao quadrado. O `LEFT JOIN` não é decorativo: um nó que perca a última aresta desaparece da agregação, e sem ele ficaria com o grau antigo para sempre — a marca de hub colada a quem já não tem ligação nenhuma. **Insiste depois de um impasse.** Isto é uma actualização em massa sobre `registry_entities`, e três trabalhos chamam-na: o `registry_content`, o `registry_expand` e o backfill nocturno. Quando duas correm ao mesmo tempo tocam as mesmas linhas em ordens diferentes e o Postgres mata uma delas — matava com ela a corrida inteira, que só ia embora fazer o trabalho da recolha outra vez uma hora depois. Um recálculo de manutenção é repetível por definição: recomeça-se no ponto de salvaguarda e tenta-se de novo. """ where = "" params: dict[str, Any] = {} if entity_ids: where = "WHERE e.id = ANY(CAST(:ids AS UUID[]))" params["ids"] = entity_ids sql = text( f""" WITH pontas AS ( SELECT holder_id AS id FROM registry_edges WHERE is_current UNION ALL SELECT subject_id FROM registry_edges WHERE is_current ), graus AS ( SELECT id, count(*) AS deg FROM pontas GROUP BY id ) UPDATE registry_entities e SET degree_current = COALESCE(g.deg, 0) FROM registry_entities e2 LEFT JOIN graus g ON g.id = e2.id WHERE e.id = e2.id AND e.degree_current IS DISTINCT FROM COALESCE(g.deg, 0) {where.replace("WHERE", "AND") if where else ""} """ ) for tentativa in range(3): try: # O ponto de salvaguarda é o que permite a repetição: sem ele o # impasse aborta a transacção do chamador e não há como continuar. async with session.begin_nested(): mudadas = (await session.execute(sql, params)).rowcount or 0 return mudadas except DBAPIError as erro: if not _e_impasse(erro) or tentativa == 2: raise logger.warning("graus: impasse, a repetir (%d/3)", tentativa + 2) await asyncio.sleep(1.0 + tentativa) return 0 async def wipeout_candidates(session: AsyncSession, limit: int = 50) -> list[dict[str, Any]]: """Empresas cuja cap table actual não fecha o capital do acto. A parte que a fonte lista sem titular **não** conta como falha: é conhecida, está quantificada em `captable_unidentified` e aparece na ficha como uma linha própria. O que esta vigia procura é o que sobra depois disso — capital que não se explica nem por titulares nem por quotas anónimas, e que portanto é parse incompleto ou estrutura parcial. """ rows = ( await session.execute( text( """ SELECT e.nif, e.name, e.captable_as_of, e.captable_total, e.captable_unidentified, a.parse_capital, a.act_type, a.id::text AS act_id, abs(e.captable_total + COALESCE(e.captable_unidentified, 0) - a.parse_capital) AS por_explicar FROM registry_entities e JOIN registry_acts a ON a.id = e.captable_act_id WHERE e.captable_total IS NOT NULL AND a.parse_capital IS NOT NULL AND abs(e.captable_total + COALESCE(e.captable_unidentified, 0) - a.parse_capital) > 0.02 ORDER BY por_explicar DESC LIMIT :lim """ ), {"lim": limit}, ) ).mappings().all() return [dict(r) for r in rows] async def parser_health(session: AsyncSession) -> dict[str, Any]: """Sinais de parser morto. O defeito que motivou toda esta fase — uma regex que exigia `NIF/NIPC:` colado ao fim de linha — devolveu zero durante meses sem nada acusar. Daí `parse_quotas` ser INT: "parseei e não encontrei" tem de se distinguir de "nunca parseei", e é a primeira que tem de disparar alarme. """ # O denominador tem de ser "actos onde a secção existe mesmo no texto", e não # "actos com corpo". Medido em produção: de 36 constituições sem quotas, 21 # tinham corpo truncado (<400 caracteres, o conteúdo não chegou a vir) e 14 # não tinham secção de quotas nenhuma — sociedades sem quotas existem. Só 1 # era candidata a falha real. Com o denominador errado o alarme dava 23% e # disparava sempre, e um alarme que dispara sempre é ignorado — que é # exactamente a falha silenciosa que isto existe para apanhar. row = ( await session.execute( text( """ SELECT count(*) FILTER (WHERE body IS NOT NULL) AS com_corpo, count(*) FILTER (WHERE body IS NOT NULL AND length(body) < 400) AS corpos_truncados, count(*) FILTER (WHERE body IS NOT NULL AND parse_version = 0) AS por_parsear, count(*) FILTER (WHERE act_class = 'constituicao' AND body ILIKE '%%QUOTA%%') AS constituicoes, count(*) FILTER (WHERE act_class = 'constituicao' AND body ILIKE '%%QUOTA%%' AND parse_quotas = 0) AS constituicoes_vazias, count(*) FILTER (WHERE act_class IN ('designacao','cessacao') AND body ILIKE '%%NIF/NIPC%%') AS actos_orgaos, count(*) FILTER (WHERE act_class IN ('designacao','cessacao') AND body ILIKE '%%NIF/NIPC%%' AND parse_people = 0) AS orgaos_vazios, count(*) FILTER (WHERE parse_error IS NOT NULL) AS com_nota FROM registry_acts """ ) ) ).mappings().first() stats = dict(row) if row else {} for total_key, empty_key, name in ( ("constituicoes", "constituicoes_vazias", "constituicao"), ("actos_orgaos", "orgaos_vazios", "orgaos"), ): total = stats.get(total_key) or 0 empty = stats.get(empty_key) or 0 stats[f"{name}_taxa_vazia"] = round(empty / total, 3) if total else None # Agora que o denominador são só os actos cuja secção existe no texto, o # limiar pode ser apertado: um acto que diga "QUOTA" e não dê nenhum titular é # falha, não feitio da fonte. 5 % dá margem para casos estranhos sem deixar # passar uma regex morta. stats["alarme"] = bool( (stats.get("constituicao_taxa_vazia") or 0) > 0.05 or (stats.get("orgaos_taxa_vazia") or 0) > 0.05 ) return stats async def suggest_identity_links(session: AsyncSession, limit: int = 500) -> int: """Candidatos a fundir um nó sem NIF com o nó com NIF da mesma pessoa. O problema é da fonte: o bloco de quotas de muitos actos está em prosa e não traz NIF nenhum, enquanto a secção dos órgãos sociais do **mesmo acto** dá o NIF de cada membro. A pessoa fica com dois nós, e o que detém a quota não leva a lado nenhum. Três evidências, por ordem de força: * **`nome+empresa`** (0.95) — nome exacto e as duas pontas ligadas à mesma empresa. É o caso da GARCIA,GARCIA & CA, onde quatro dos cinco sócios sem NIF são administradores com NIF da própria empresa; * **`nome+rua+cp`** (0.90) — nome exacto e a **morada completa** igual. Medido em 5.356 pares: a morada tem 45 caracteres em média e 5.348 deles trazem número de porta ou lote. Não é "a mesma rua", é a mesma casa; * **`nome+cp`** (0.60) — código postal sem rua. Fica sempre para revisão: um código postal chega para um bairro inteiro. As duas primeiras confirmam-se sozinhas, **mas só quando o nó sem NIF tem um único candidato**. Com dois, escolher seria tirar à sorte, e esses continuam a precisar de uma pessoa — são 3 nós em 5.353. **Só nome não gera sugestão.** São 10.323 pares no índice e homónimos foi exactamente o que estragou as relações antigas — uma lista dessas não se revê, ignora-se. """ # A mesma empresa nas duas pontas é o sinal que dispensa uma pessoa a # decidir, e por isso é o único que entra já confirmado. `confirmed_by` nulo # marca-o como automático, para se poder auditar e desfazer. result = await session.execute( text( """ WITH pares AS ( SELECT DISTINCT u.id AS unknown_id, c.id AS candidate_id, CASE WHEN EXISTS ( SELECT 1 FROM registry_edges r1 JOIN registry_edges r2 ON r2.subject_id = r1.subject_id WHERE r1.holder_id = u.id AND r2.holder_id = c.id ) THEN 'nome+empresa' WHEN u.address_norm IS NOT NULL AND u.address_norm = c.address_norm THEN 'nome+rua+cp' WHEN u.codigo_postal IS NOT NULL AND u.codigo_postal = c.codigo_postal THEN 'nome+cp' END AS evidence FROM registry_entities u JOIN registry_entities c ON c.name_norm = u.name_norm AND c.nif IS NOT NULL AND c.kind = 'person' AND c.id <> u.id WHERE u.nif IS NULL AND u.merged_into IS NULL AND length(u.name_norm) >= 10 ), contadas AS ( SELECT p.*, count(*) OVER (PARTITION BY p.unknown_id) AS candidatos FROM pares p WHERE p.evidence IS NOT NULL ) INSERT INTO registry_identity_links (unknown_id, candidate_id, evidence, score, confirmed_at) SELECT unknown_id, candidate_id, evidence, CASE evidence WHEN 'nome+empresa' THEN 0.95 WHEN 'nome+rua+cp' THEN 0.90 ELSE 0.60 END, CASE WHEN candidatos = 1 AND evidence IN ('nome+empresa', 'nome+rua+cp') THEN now() END FROM contadas LIMIT :lim ON CONFLICT (unknown_id, candidate_id) DO NOTHING """ ), {"lim": limit}, ) total = result.rowcount or 0 # Os pares já sugeridos por corridas anteriores não voltam a passar pelo # INSERT — o `ON CONFLICT DO NOTHING` salta-os —, e ficariam com a evidência # fraca com que nasceram. Medido: dos 531 candidatos que já existiam, 157 # tinham as duas pontas na mesma empresa e continuavam à espera de alguém. promovidos = await session.execute( text( """ UPDATE registry_identity_links l SET evidence = 'nome+empresa', score = 0.95, confirmed_at = now() WHERE l.confirmed_at IS NULL AND l.rejected_at IS NULL AND l.evidence <> 'nome+empresa' AND EXISTS ( SELECT 1 FROM registry_edges r1 JOIN registry_edges r2 ON r2.subject_id = r1.subject_id WHERE r1.holder_id = l.unknown_id AND r2.holder_id = l.candidate_id ) """ ) ) # E os `nome+rua+cp` que já lá estavam nasceram à espera de revisão, porque # antes só a empresa se confirmava sozinha. Eram 5.356 numa fila que ninguém # ia rever à mão. A condição de não haver outro candidato é a mesma do INSERT # — é ela que separa "é obviamente a mesma pessoa" de "é uma de duas". morada = await session.execute( text( """ UPDATE registry_identity_links l SET confirmed_at = now() WHERE l.confirmed_at IS NULL AND l.rejected_at IS NULL AND l.evidence = 'nome+rua+cp' AND NOT EXISTS ( SELECT 1 FROM registry_identity_links o WHERE o.unknown_id = l.unknown_id AND o.id <> l.id AND o.rejected_at IS NULL ) """ ) ) return total + (promovidos.rowcount or 0) + (morada.rowcount or 0) async def apply_identity_merges(session: AsyncSession) -> dict[str, int]: """Aponta as arestas dos nós absorvidos para a identidade confirmada. Não apaga nada. O nó sem NIF continua a existir com o nome que o acto publicou — é um facto da fonte, e é o que permite desfazer — mas as arestas dele passam a sair do nó com NIF, que é a mesma pessoa. É isto que liga a quota de 235.000 € da Teresa Maria ao resto do grafo em vez de a deixar num beco. Uma aresta que ao ser reapontada colidisse com uma já existente (o mesmo acto, o mesmo par, o mesmo tipo) é apagada em vez de reapontada: seria a mesma afirmação duas vezes. """ out = {"nos": 0, "arestas": 0, "duplicadas": 0} absorventes: set[str] = set() pares = ( await session.execute( text( """ SELECT l.unknown_id::text, l.candidate_id::text FROM registry_identity_links l JOIN registry_entities u ON u.id = l.unknown_id WHERE l.confirmed_at IS NOT NULL AND l.rejected_at IS NULL AND u.merged_into IS NULL """ ) ) ).all() for unknown_id, candidate_id in pares: dup = await session.execute( text( """ DELETE FROM registry_edges e WHERE e.holder_id = CAST(:u AS UUID) AND EXISTS ( SELECT 1 FROM registry_edges o WHERE o.source_act_id = e.source_act_id AND o.holder_id = CAST(:c AS UUID) AND o.subject_id = e.subject_id AND o.edge_type = e.edge_type AND o.snapshot_seq = e.snapshot_seq ) """ ), {"u": unknown_id, "c": candidate_id}, ) out["duplicadas"] += dup.rowcount or 0 moved = await session.execute( text( """ UPDATE registry_edges SET holder_id = CAST(:c AS UUID) WHERE holder_id = CAST(:u AS UUID) AND subject_id <> CAST(:c AS UUID) """ ), {"u": unknown_id, "c": candidate_id}, ) out["arestas"] += moved.rowcount or 0 await session.execute( text( "UPDATE registry_entities SET merged_into = CAST(:c AS UUID)," " updated_at = now() WHERE id = CAST(:u AS UUID)" ), {"u": unknown_id, "c": candidate_id}, ) out["nos"] += 1 absorventes.add(candidate_id) if out["nos"]: # As quotas mudaram de mão: as percentagens e o `is_current` de cada # empresa tocada têm de ser recalculados, e o grau de todos. # # **Só as empresas dos nós absorvidos nesta corrida.** A versão anterior # partia de `u.merged_into IS NOT NULL` — todos os nós já fundidos desde # sempre — e recalculava 14.137 empresas de cada vez que aparecesse uma # única fusão nova. Além do desperdício, era a consulta que rebentava o # `/dev/shm` de 64 MB e derrubava a corrida inteira do `registry_content` # hora a hora, com as fusões a nunca chegarem a ser aplicadas. afectadas = ( await session.execute( text( """ SELECT DISTINCT s.id::text, s.nif FROM registry_edges r JOIN registry_entities s ON s.id = r.subject_id WHERE r.holder_id = ANY(CAST(:ids AS UUID[])) AND s.nif IS NOT NULL """ ), {"ids": sorted(absorventes)}, ) ).all() for subject_id, nipc in afectadas: await recompute_current(session, subject_id, nipc) await refresh_degrees(session) logger.info("fusões de identidade: %s", out) return out