"""Escrita do corpus de actos e das arestas do grafo. Fluxo de um acto: `upsert_act` guarda a linha (com ou sem corpo) → `apply_parse` lê o resultado do `registry_act_parser`, cria os nós, substitui as arestas desse acto e manda o `registry_graph.recompute_current` reavaliar o presente. As arestas são **afirmações de um acto**, não estado. É essa escolha que torna o reparse trivial: apagam-se as arestas de `source_act_id`, reinserem-se, e o estado actual recalcula-se. Sem isso, melhorar uma regex obrigaria a refazer a recolha — e refazer um ano de conteúdo custa ~14 h de HTTP. """ import hashlib import json import logging from datetime import date, timedelta from typing import Any from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.services.registry_act_parser import ( CAPTABLE_CLASSES, REGISTRY_PARSE_VERSION, classify_act, entity_key, entity_kind, normalize_address, normalize_name, wants_content, ) from app.services.registry_graph import recompute_current logger = logging.getLogger(__name__) # Cargo -> tipo de aresta. O texto do cargo guarda-se sempre em `role_label`; o # tipo serve para a travessia poder pedir "só propriedade" ou "só gestão". _EDGE_BY_ROLE: tuple[tuple[str, tuple[str, ...]], ...] = ( ("audits", ("fiscal", "roc", "sroc", "revisor", "auditor")), ("secretary", ("secretári", "secretari")), ("liquidator", ("liquidat",)), ("depositary", ("deposit",)), ("manages", ( "gerent", "gerênc", "gerenc", "administra", "presiden", "vogal", "conselho", "direcç", "direc", "diretor", "director", "representante", "mesa da assembleia", "suplente", )), ) def edge_type_for(cargo: str | None, organ: str | None = None) -> str: haystack = f"{cargo or ''} {organ or ''}".lower() for edge_type, needles in _EDGE_BY_ROLE: if any(n in haystack for n in needles): return edge_type return "other" # Um acto de rectificação reimprime uma inscrição anterior. Sem preâmbulo "pela # Apresentação" não há a apresentação do próprio acto para comparar, e resta o # `act_date` — mas entre o registo e a publicação passam-se dias ou semanas em # actos perfeitamente normais, e confundir esse atraso com uma republicação # mudaria a data de dezenas de milhares de factos. Daí a folga. _REPUBLICATION_SLACK = timedelta(days=90) def _fact_date(ap: date | None, own_ap: date | None, act_date: date | None) -> date | None: """A que data pertence um facto: à do bloco onde está, ou à do acto. O bloco traz a sua própria apresentação. Num acto normal é a mesma do acto e não há nada a decidir; num "Rectificado" ou "Actualizado" é a da inscrição reimpressa, e é essa que vale. Foi o que pôs o Luís Miguel de Sousa Leal a administrar a IGPS PROTEK a partir de 2015 com uma designação de 2008, cinco anos depois de ter renunciado. """ if ap is None or act_date is None: return act_date if own_ap is not None: return ap if ap < own_ap else act_date return ap if act_date - ap > _REPUBLICATION_SLACK else act_date def _parse_note(parsed: dict[str, Any]) -> str | None: """Porque é que este acto não serve como estrutura actual, se for o caso. Fica gravado em `parse_error` — que aqui é mais nota do que erro. O que interessa é não descobrir só mais tarde, ao olhar para uma ficha estranha, que o acto tinha uma limitação conhecida à partida. """ quotas = parsed.get("quotas") or [] if not quotas: return None notes: list[str] = [] orphans = (parsed.get("counts") or {}).get("orphans", 0) if orphans: notes.append( f"{orphans} quota(s) listada(s) sem titular na fonte: não serve como " "estrutura actual" ) if not parsed.get("captable_linkable"): notes.append("cap table sem NIF em nenhum titular (formato antigo em prosa)") if len(parsed.get("quota_currencies") or []) > 1: notes.append("quotas em moedas diferentes: total não somado") return "; ".join(notes) or None def act_dedup_hash(nipc: str, irn_id: str | None, fallback: str | None = None) -> str: """Mesma receita do `mj_irn_sync._dedup`, de propósito. O backfill do que já está em `dre_publications` e o varrimento nacional vão encontrar o mesmo acto por caminhos diferentes; se as chaves divergissem, ficávamos com duas linhas para o mesmo facto e duas cap tables a competir. """ key = irn_id or fallback or "" return hashlib.sha256(f"mj|{nipc}|{key}".encode()).hexdigest() async def upsert_entity( session: AsyncSession, *, nif: str | None, name: str | None, header: dict[str, Any] | None = None, address_norm: str | None = None, codigo_postal: str | None = None, pais: str | None = None, act_date: date | None = None, ) -> str | None: """Cria ou actualiza um nó. Devolve o id, ou None se não houver identidade. **Duas regras para o nome, e a diferença é quem o está a dizer.** Quando o acto é *sobre* esta entidade — vem com `header`, portanto traz a firma tal como o registo a publica —, o nome do acto mais recente ganha. Uma empresa muda de firma, e transformar-se em anónima é precisamente uma das alturas em que isso acontece: a JOIN THE MOMENT TRANSITÁRIOS passou a `JOIN THE MOMENT - TRANSITÁRIOS, S.A.` em 2022 e continuava guardada como `... UNIPESSOAL LDA` porque o nome novo tem 35 caracteres e o antigo 43. Quando a entidade aparece só como parte no acto de outra — sem `header` — mantém-se o critério antigo, o mais comprido: aí a fonte alterna entre `CRESCENTIMOB - IMOBILIÁRIA, LDA` e `Crescentimob`, e a forma completa é a útil. Deixar a versão abreviada de um acto recente ganhar seria trocar a firma por uma alcunha. O cabeçalho do registo só se escreve quando vem preenchido — um acto que não traga a sede não deve apagar a que lá estava. """ key = entity_key(nif, name) if not key: return None header = header or {} display = (name or header.get("firma") or nif or "").strip()[:300] if not display: return None row = ( await session.execute( text( """ INSERT INTO registry_entities ( nif, entity_key, nif_valid, kind, name, name_norm, natureza_juridica, sede, distrito, concelho, freguesia, codigo_postal, localidade, pais, address_norm, last_act_date, header_as_of ) VALUES ( :nif, :key, :nif_valid, :kind, :name, :name_norm, :natureza, :sede, :distrito, :concelho, :freguesia, :cp, :localidade, :pais, :addr, CAST(:adate AS DATE), CASE WHEN :tem_cabecalho THEN CAST(:adate AS DATE) END ) ON CONFLICT (entity_key) DO UPDATE SET name = CASE WHEN :firma_propria AND EXCLUDED.last_act_date IS NOT NULL AND (registry_entities.header_as_of IS NULL OR EXCLUDED.last_act_date >= registry_entities.header_as_of) THEN EXCLUDED.name WHEN length(EXCLUDED.name) > length(registry_entities.name) THEN EXCLUDED.name ELSE registry_entities.name END, name_norm = CASE WHEN :firma_propria AND EXCLUDED.last_act_date IS NOT NULL AND (registry_entities.header_as_of IS NULL OR EXCLUDED.last_act_date >= registry_entities.header_as_of) THEN EXCLUDED.name_norm WHEN length(EXCLUDED.name) > length(registry_entities.name) THEN EXCLUDED.name_norm ELSE registry_entities.name_norm END, -- O cabeçalho inteiro segue a mesma regra do nome: vale o do -- acto mais recente. Com `COALESCE` ganhava o **último a ser -- ingerido**, e os actos entram em ordem arbitrária — a FORD -- LUSITANA ficou guardada como Sociedade Anónima com base num -- acto de 2020, tendo três actos posteriores a dizer -- Sociedade por Quotas. Uma empresa muda de forma, de sede e -- de concelho; o que não pode é o retrato depender de quem -- chegou primeiro à base de dados. natureza_juridica = CASE WHEN EXCLUDED.natureza_juridica IS NOT NULL AND EXCLUDED.last_act_date IS NOT NULL AND (registry_entities.header_as_of IS NULL OR EXCLUDED.last_act_date >= registry_entities.header_as_of) THEN EXCLUDED.natureza_juridica ELSE COALESCE(registry_entities.natureza_juridica, EXCLUDED.natureza_juridica) END, sede = CASE WHEN EXCLUDED.sede IS NOT NULL AND EXCLUDED.last_act_date IS NOT NULL AND (registry_entities.header_as_of IS NULL OR EXCLUDED.last_act_date >= registry_entities.header_as_of) THEN EXCLUDED.sede ELSE COALESCE(registry_entities.sede, EXCLUDED.sede) END, distrito = CASE WHEN EXCLUDED.distrito IS NOT NULL AND EXCLUDED.last_act_date IS NOT NULL AND (registry_entities.header_as_of IS NULL OR EXCLUDED.last_act_date >= registry_entities.header_as_of) THEN EXCLUDED.distrito ELSE COALESCE(registry_entities.distrito, EXCLUDED.distrito) END, concelho = CASE WHEN EXCLUDED.concelho IS NOT NULL AND EXCLUDED.last_act_date IS NOT NULL AND (registry_entities.header_as_of IS NULL OR EXCLUDED.last_act_date >= registry_entities.header_as_of) THEN EXCLUDED.concelho ELSE COALESCE(registry_entities.concelho, EXCLUDED.concelho) END, freguesia = CASE WHEN EXCLUDED.freguesia IS NOT NULL AND EXCLUDED.last_act_date IS NOT NULL AND (registry_entities.header_as_of IS NULL OR EXCLUDED.last_act_date >= registry_entities.header_as_of) THEN EXCLUDED.freguesia ELSE COALESCE(registry_entities.freguesia, EXCLUDED.freguesia) END, codigo_postal = COALESCE(EXCLUDED.codigo_postal, registry_entities.codigo_postal), localidade = COALESCE(EXCLUDED.localidade, registry_entities.localidade), pais = COALESCE(EXCLUDED.pais, registry_entities.pais), address_norm = COALESCE(EXCLUDED.address_norm, registry_entities.address_norm), last_act_date = GREATEST( registry_entities.last_act_date, COALESCE(EXCLUDED.last_act_date, registry_entities.last_act_date)), -- A data do cabeçalho que ficou guardado. Sem ela, a -- comparação era feita contra o `last_act_date`, que conta -- actos nunca lidos: uma empresa cuja publicação mais recente -- é uma prestação de contas ficava presa ao primeiro -- cabeçalho que por acaso lá tivesse chegado. header_as_of = CASE WHEN :tem_cabecalho AND EXCLUDED.last_act_date IS NOT NULL AND (registry_entities.header_as_of IS NULL OR EXCLUDED.last_act_date >= registry_entities.header_as_of) THEN EXCLUDED.last_act_date ELSE registry_entities.header_as_of END, updated_at = now() RETURNING id::text """ ), { "nif": nif, "key": key, "nif_valid": entity_kind(nif) not in ("unknown", "invalid"), "kind": entity_kind(nif), "name": display, "name_norm": normalize_name(display), # O `header` só chega quando o acto é sobre esta entidade. "firma_propria": bool(header.get("firma")), "tem_cabecalho": bool(header), "natureza": header.get("natureza_juridica"), "sede": header.get("sede"), "distrito": header.get("distrito"), "concelho": header.get("concelho"), "freguesia": header.get("freguesia"), "cp": codigo_postal or header.get("codigo_postal"), "localidade": header.get("localidade"), "pais": pais, "addr": address_norm, "adate": act_date, }, ) ).first() if not row: return None if nif: return row[0] # Sem NIF, o nó pode já ter sido reconhecido como sendo a mesma pessoa de um # nó com NIF. As arestas deste acto saem dessa identidade, senão cada reparse # desfazia a fusão em silêncio — o `apply_parse` apaga as arestas do acto e # reinsere-as, e reinseri-las no nó absorvido punha a quota outra vez num beco. merged = ( await session.execute( text("SELECT merged_into::text FROM registry_entities WHERE id = CAST(:id AS UUID)"), {"id": row[0]}, ) ).scalar() return merged or row[0] async def upsert_act( session: AsyncSession, *, nipc: str, irn_id: str | None, act_type: str | None, act_date: date | None, entity_name: str | None = None, listing: dict[str, Any] | None = None, body: str | None = None, body_source: str = "irn", body_truncated: bool = False, district: str | None = None, council: str | None = None, apresentacao: str | None = None, priority: int = 50, ) -> tuple[str, bool]: """Grava o acto. Devolve (id, é_novo). Quando já existe, só se acrescenta: um corpo novo substitui a ausência de corpo, mas nunca se apaga um corpo por passar uma listagem sem ele. É o que permite ao varrimento nacional e ao histórico por NIPC cruzarem-se sem se estragarem. """ act_class = classify_act(act_type) row = ( await session.execute( text( """ INSERT INTO registry_acts ( nipc, irn_publication_id, entity_name, act_type, act_class, act_date, apresentacao, district, council, listing_json, body, body_sha256, body_source, body_truncated, content_status, wants_content, priority, dedup_hash ) VALUES ( :nipc, :irn_id, :entity_name, :act_type, :act_class, CAST(:act_date AS DATE), :apresentacao, :district, :council, CAST(:listing AS JSONB), :body, :sha, :body_source, :truncated, :status, :wants, :priority, :hash ) ON CONFLICT (dedup_hash) DO UPDATE SET entity_name = COALESCE(EXCLUDED.entity_name, registry_acts.entity_name), act_type = COALESCE(EXCLUDED.act_type, registry_acts.act_type), act_class = CASE WHEN registry_acts.act_class = 'outro' THEN EXCLUDED.act_class ELSE registry_acts.act_class END, act_date = COALESCE(EXCLUDED.act_date, registry_acts.act_date), apresentacao = COALESCE(EXCLUDED.apresentacao, registry_acts.apresentacao), district = COALESCE(EXCLUDED.district, registry_acts.district), council = COALESCE(EXCLUDED.council, registry_acts.council), listing_json = COALESCE(EXCLUDED.listing_json, registry_acts.listing_json), body = COALESCE(EXCLUDED.body, registry_acts.body), body_sha256 = COALESCE(EXCLUDED.body_sha256, registry_acts.body_sha256), body_truncated = registry_acts.body_truncated AND EXCLUDED.body_truncated, content_status = CASE WHEN EXCLUDED.body IS NOT NULL THEN 'fetched' ELSE registry_acts.content_status END, content_fetched_at = CASE WHEN EXCLUDED.body IS NOT NULL THEN now() ELSE registry_acts.content_fetched_at END, priority = LEAST(registry_acts.priority, EXCLUDED.priority), last_seen_at = now() RETURNING id::text, (xmax = 0) AS inserted """ ), { "nipc": nipc, "irn_id": str(irn_id) if irn_id is not None else None, "entity_name": (entity_name or None), "act_type": act_type, "act_class": act_class, "act_date": act_date, "apresentacao": apresentacao, "district": district, "council": council, "listing": json.dumps(listing, ensure_ascii=False) if listing else None, "body": body, "sha": hashlib.sha256(body.encode()).hexdigest() if body else None, "body_source": body_source, "truncated": body_truncated, "status": "fetched" if body else "pending", "wants": wants_content(act_type), "priority": priority, "hash": act_dedup_hash(nipc, irn_id, act_type), }, ) ).first() return (row[0], bool(row[1])) if row else ("", False) async def _spouse_edge( session: AsyncSession, *, act_id: str, holder_of_spouse: str, person: dict[str, Any], act_date: date | None, ) -> int: """Aresta `spouse_of` a partir do nome do cônjuge. O cônjuge vem sem NIF, portanto o nó é chaveado por nome e **nunca** se funde com um nó com NIF automaticamente. O que o torna útil é herdar a morada do titular: nome igual mais morada igual é a evidência que a `registry_graph.suggest_identity_links` transforma numa sugestão de confiança alta, para alguém confirmar. """ conjuge = person.get("conjuge") if not conjuge: return 0 spouse_id = await upsert_entity( session, nif=None, name=conjuge, address_norm=person.get("address_norm"), codigo_postal=person.get("codigo_postal"), act_date=act_date, ) if not spouse_id or spouse_id == holder_of_spouse: return 0 await session.execute( text( """ INSERT INTO registry_edges ( holder_id, subject_id, edge_type, role_label, valid_from, is_current, source_act_id, snapshot_seq, confidence, parse_version ) VALUES ( CAST(:holder AS UUID), CAST(:subject AS UUID), 'spouse_of', :regime, CAST(:adate AS DATE), TRUE, CAST(:aid AS UUID), 0, 0.95, :pv ) ON CONFLICT (source_act_id, holder_id, subject_id, edge_type, snapshot_seq) DO NOTHING """ ), { "holder": spouse_id, "subject": holder_of_spouse, "regime": (person.get("regime_bens") or None), "adate": act_date, "aid": act_id, "pv": REGISTRY_PARSE_VERSION, }, ) return 1 def _orphan_total(parsed: dict[str, Any]) -> float | None: """Soma das quotas que a fonte lista sem nomear titular. Só se soma quando **todas** têm valor e a moeda é uma só, pela mesma razão que o parser aplica ao total das quotas: um total parcial apresentado como se fosse completo é pior do que não haver total nenhum. """ orphans = parsed.get("quota_orphans") or [] if not orphans or any(o.get("quota_amount") is None for o in orphans): return None if len({o.get("quota_currency") for o in orphans if o.get("quota_currency")}) > 1: return None return round(sum(o["quota_amount"] for o in orphans), 2) async def apply_parse( session: AsyncSession, *, act_id: str, nipc: str, parsed: dict[str, Any], act_date: date | None, ) -> dict[str, int]: """Cria nós e arestas de um acto já parseado, e reavalia o presente. Apaga primeiro as arestas deste acto: é o que faz o reparse ser idempotente. Nenhuma outra rotina apaga arestas. """ counts = {"quotas": 0, "officers": 0, "spouses": 0, "skipped": 0} header = parsed.get("header") or {} act_class = parsed.get("act_class") or "outro" is_cessation = act_class == "cessacao" own_ap = parsed.get("act_own_ap") # A data mais antiga que este acto reimprime, se reimprimir alguma. Fica # gravada em `parse_effective_date` para a escolha da estrutura societária não # tomar uma cap table republicada por ser a mais recente. reprinted: list[date] = [] subject_id = await upsert_entity( session, nif=nipc, name=header.get("firma") or parsed.get("entity_name"), header=header, act_date=act_date, ) if not subject_id: return counts await session.execute( text("DELETE FROM registry_edges WHERE source_act_id = CAST(:aid AS UUID)"), {"aid": act_id}, ) for quota in parsed.get("quotas") or []: # "AMORTIZADAS" não é um sócio: é o estado da quota, que pertence à # sociedade. Conta para o total (e é por isso que a soma bate com o # capital) mas não pode virar uma pessoa. if quota.get("is_own_quota"): counts["skipped"] += 1 continue qdate = _fact_date(quota.get("ap_date"), own_ap, act_date) if qdate != act_date and qdate is not None: reprinted.append(qdate) holder_id = await upsert_entity( session, nif=quota.get("nif"), name=quota.get("name"), address_norm=quota.get("address_norm"), codigo_postal=quota.get("codigo_postal"), pais=quota.get("pais"), act_date=qdate, ) if not holder_id or holder_id == subject_id: counts["skipped"] += 1 continue await session.execute( text( """ INSERT INTO registry_edges ( holder_id, subject_id, edge_type, role_label, quota_amount, currency, valid_from, is_current, source_act_id, snapshot_seq, confidence, parse_version ) VALUES ( CAST(:holder AS UUID), CAST(:subject AS UUID), 'holds_quota', 'sócio', :amount, :currency, CAST(:adate AS DATE), TRUE, CAST(:aid AS UUID), :seq, :conf, :pv ) ON CONFLICT (source_act_id, holder_id, subject_id, edge_type, snapshot_seq) DO NOTHING """ ), { "holder": holder_id, "subject": subject_id, "amount": quota.get("quota_amount"), "currency": quota.get("quota_currency"), "adate": qdate, "aid": act_id, "seq": quota.get("seq") or 0, # sem NIF a aresta continua a valer para a ficha da empresa, mas # entra com confiança menor e o nó não se segue no grafo "conf": 1.00 if quota.get("nif") else 0.50, "pv": REGISTRY_PARSE_VERSION, }, ) counts["quotas"] += 1 counts["spouses"] += await _spouse_edge( session, act_id=act_id, holder_of_spouse=holder_id, person=quota, act_date=qdate, ) for seq, officer in enumerate(parsed.get("officers") or []): odate = _fact_date(officer.get("ap_date"), own_ap, act_date) if odate != act_date and odate is not None: reprinted.append(odate) # A natureza é a do bloco onde a pessoa foi encontrada; o `act_class` do # acto inteiro é só o recurso para corpos sem blocos. ceased = officer.get("cessation") if ceased is None: ceased = is_cessation holder_id = await upsert_entity( session, nif=officer.get("nif"), name=officer.get("name"), address_norm=officer.get("address_norm"), codigo_postal=officer.get("codigo_postal"), act_date=odate, ) if not holder_id or holder_id == subject_id: counts["skipped"] += 1 continue await session.execute( text( """ INSERT INTO registry_edges ( holder_id, subject_id, edge_type, role_label, valid_from, valid_to, is_current, cease_cause, source_act_id, snapshot_seq, confidence, parse_version ) VALUES ( CAST(:holder AS UUID), CAST(:subject AS UUID), :etype, :role, CAST(:adate AS DATE), CAST(:vto AS DATE), :current, :causa, CAST(:aid AS UUID), :seq, :conf, :pv ) ON CONFLICT (source_act_id, holder_id, subject_id, edge_type, snapshot_seq) DO NOTHING """ ), { "holder": holder_id, "subject": subject_id, "etype": edge_type_for(officer.get("cargo"), officer.get("organ")), "role": (officer.get("cargo") or officer.get("organ") or None), "adate": odate, # uma cessação nasce já fechada, com a data do acto "vto": odate if ceased else None, "current": not ceased, "causa": officer.get("causa"), "aid": act_id, "seq": seq, "conf": 1.00 if officer.get("nif") else 0.50, "pv": REGISTRY_PARSE_VERSION, }, ) counts["officers"] += 1 counts["spouses"] += await _spouse_edge( session, act_id=act_id, holder_of_spouse=holder_id, person=officer, act_date=odate, ) await session.execute( text( """ UPDATE registry_acts SET parse_version = :pv, parsed_at = now(), parse_quotas = :nq, parse_people = :np, parse_quota_orphans = :norph, parse_quota_orphan_total = :orph_total, parse_capital = :capital, parse_captable_complete = :complete, parse_captable_eligible = :eligible, parse_effective_date = CAST(:effective AS DATE), parse_error = :err WHERE id = CAST(:aid AS UUID) """ ), { "pv": REGISTRY_PARSE_VERSION, "nq": counts["quotas"], "np": counts["officers"], "norph": (parsed.get("counts") or {}).get("orphans", 0), "orph_total": _orphan_total(parsed), "capital": parsed.get("capital"), "complete": parsed.get("captable_complete"), "eligible": parsed.get("captable_eligible"), # Fica NULL no caso normal — é o que permite a `COALESCE` na escolha da # estrutura societária tratar os dois casos com a mesma consulta. "effective": min(reprinted) if reprinted else None, # A ausência de NIF numa cap table é a marca das publicações de 2006 # em prosa. Fica registada para se saber que aquele acto vale para a # ficha mas não para o grafo, em vez de se descobrir depois. "err": _parse_note(parsed), "aid": act_id, }, ) await session.execute( text( """ UPDATE registry_entities SET acts_count = (SELECT count(*) FROM registry_acts WHERE nipc = :nipc), updated_at = now() WHERE id = CAST(:sid AS UUID) """ ), {"nipc": nipc, "sid": subject_id}, ) await recompute_current(session, subject_id, nipc) return counts async def enqueue_entity( session: AsyncSession, *, nif: str, reason: str, depth: int = 0, seed_nif: str | None = None, ) -> bool: """Põe um NIPC na fila do histórico por NIPC. Devolve True se entrou agora. Não desce a profundidade de uma entrada que já esteja mais perto do nosso universo: um NIPC pedido como cliente (depth 0) não deve passar a depth 2 por aparecer também como sócio de um sócio. """ row = ( await session.execute( text( """ INSERT INTO registry_entity_fetch (nif, depth, reason, seed_nif) VALUES (:nif, :depth, :reason, :seed) ON CONFLICT (nif) DO UPDATE SET depth = LEAST(registry_entity_fetch.depth, EXCLUDED.depth), reason = CASE WHEN EXCLUDED.depth < registry_entity_fetch.depth THEN EXCLUDED.reason ELSE registry_entity_fetch.reason END RETURNING (xmax = 0) AS inserted """ ), {"nif": nif, "depth": depth, "reason": reason, "seed": seed_nif}, ) ).first() return bool(row and row[0]) async def link_monitored_companies(session: AsyncSession) -> int: """Marca os nós que também são empresas monitorizadas. A ligação é por NIF e não por chave estrangeira: `companies` tem cascatas em todo o lado e o nó nacional tem de sobreviver a uma empresa ser removida da monitorização. """ result = await session.execute( text( """ INSERT INTO entity_incident_summary (nif, monitored_company_id, monitoring_type, risk_score) SELECT c.nif, c.id, c.monitoring_type, c.risk_score FROM companies c ON CONFLICT (nif) DO UPDATE SET monitored_company_id = EXCLUDED.monitored_company_id, monitoring_type = EXCLUDED.monitoring_type, risk_score = EXCLUDED.risk_score, computed_at = now() """ ) ) return result.rowcount or 0 CAPTABLE_ACT_CLASSES = tuple(sorted(CAPTABLE_CLASSES))