"""Sincronização das publicações societárias por NIF, via portal IRN. Guarda em `dre_publications` com `source='mj'` — a tabela e a UI já existem desde a fase da extensão Chrome, que isto substitui. As pessoas (sócios, gerência, fiscalização) continuam a sair pelo `mj_people_parser`, que já sabia ler exactamente este formato. Duas decisões de custo: * **O conteúdo só é aberto quando é preciso.** A listagem por NIF é uma chamada; abrir o conteúdo são três. A esmagadora maioria das publicações são prestações de contas, que não têm pessoas nem alteram nada — abrimos só as que ainda não temos e as que interessam para enriquecer a ficha. * **Sem sweep nacional.** São 683 publicações/dia à escala do país; por NIF sobre as entidades monitorizadas são ~22 cada, uma vez, e depois só o novo. """ import hashlib import json import logging from datetime import date, datetime from typing import Any from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.scrapers.mj_irn import MjIrnClient from app.services.mj_company_enrichment import ( capital_is_current, html_to_text, merge_company_fields, parse_publication, select_enrichment_publications, ) from app.services.mj_people_parser import extract_people logger = logging.getLogger(__name__) # Actos que não trazem pessoas nem alteram a ficha: não vale a pena abrir o # conteúdo. São a maioria esmagadora do volume. SKIP_CONTENT_PREFIXES = ("PRESTAÇÃO DE CONTAS", "PRESTACAO DE CONTAS") def _dedup(nipc: str, pub_id: str) -> str: return hashlib.sha256(f"mj|{nipc}|{pub_id}".encode()).hexdigest() def _parse_date(s: str | None) -> date | None: if not s: return None try: return datetime.strptime(s[:10], "%Y-%m-%d").date() except ValueError: return None def _wants_content(act: str | None) -> bool: a = (act or "").strip().upper() return not a.startswith(SKIP_CONTENT_PREFIXES) async def sync_company( session: AsyncSession, client: MjIrnClient, company_id: str, nif: str ) -> dict[str, int]: """Sincroniza um NIF. Devolve contagens para o log do scrape.""" pubs = await client.publications_for_nif(nif) stats = {"seen": len(pubs), "new": 0, "people": 0} if not pubs: return stats known = { r[0] for r in ( await session.execute( text( "SELECT raw_json->>'Id' FROM dre_publications" " WHERE company_id = :cid AND source = 'mj'" ), {"cid": company_id}, ) ).all() } # Publicações escolhidas para enriquecer a ficha da empresa. Abrimos sempre, # mesmo já conhecidas, porque a ficha pode nunca ter sido preenchida. enrich_ids = {p["Id"] for p in select_enrichment_publications(pubs)} enrich_parsed: list[dict[str, Any]] = [] for pub in pubs: pub_id = str(pub.get("Id")) is_new = pub_id not in known needs_content = pub_id in enrich_ids or (is_new and _wants_content(pub.get("Act_Fact"))) html = await client.content_for(pub) if needs_content else None if pub_id in enrich_ids: enrich_parsed.append(parse_publication(html)) if not is_new: continue body = html_to_text(html) if html else None row = ( await session.execute( text( """ INSERT INTO dre_publications (company_id, title, summary, date, type, source, raw_json, dedup_hash) VALUES (:cid, :title, :summary, :date, :type, 'mj', CAST(:raw AS JSONB), :hash) ON CONFLICT (dedup_hash) DO NOTHING RETURNING id::text """ ), { "cid": company_id, "title": (pub.get("Act_Fact") or "Publicação")[:500], "summary": (body or "")[:4000] or None, "date": _parse_date(pub.get("Date")), "type": pub.get("Act_Fact"), "raw": json.dumps(pub, ensure_ascii=False), "hash": _dedup(nif, pub_id), }, ) ).first() if not row: continue stats["new"] += 1 publication_id = row[0] for person in extract_people(body): nif_p = (person.get("nif") or "").strip() if not nif_p: continue await session.execute( text( """ INSERT INTO publication_people (publication_id, company_id, person_nif, person_name, cargo, event, act_date) VALUES (:pid, :cid, :nif, :name, :cargo, :event, :adate) ON CONFLICT DO NOTHING """ ), { "pid": publication_id, "cid": company_id, "nif": nif_p, "name": person.get("name"), "cargo": person.get("cargo"), "event": person.get("event") or person.get("role_kind") or "other", "adate": _parse_date(pub.get("Date")), }, ) stats["people"] += 1 if enrich_parsed: await _apply_enrichment( session, company_id, enrich_parsed, keep_capital=capital_is_current(pubs) ) return stats async def _apply_enrichment( session: AsyncSession, company_id: str, parsed: list[dict[str, Any]], keep_capital: bool = True, ) -> None: """Grava os campos do registo comercial. `COALESCE(:campo, coluna)` em vez de sobrescrever: uma publicação que não traga um campo não deve apagar o que já lá estava. """ fields = merge_company_fields(parsed) if not keep_capital: # houve aumento/redução de capital depois da constituição: o valor # inicial já não vale e o novo não vem publicado fields["capital_social"] = None await session.execute( text( """ UPDATE companies SET legal_name = COALESCE(NULLIF(:firma,''), legal_name), natureza_juridica = COALESCE(:natureza, natureza_juridica), address = COALESCE(NULLIF(:sede,''), address), distrito = COALESCE(:distrito, distrito), concelho = COALESCE(:concelho, concelho), freguesia = COALESCE(:freguesia, freguesia), codigo_postal = COALESCE(:cp, codigo_postal), localidade = COALESCE(:localidade, localidade), objeto_social = COALESCE(:objeto, objeto_social), capital_social = COALESCE(:capital, capital_social), registry_complete = companies.registry_complete OR :completa, registry_fetched_at = now(), updated_at = now() WHERE id = :cid """ ), { "cid": company_id, "firma": fields.get("firma") or "", "natureza": fields.get("natureza_juridica"), "sede": fields.get("sede") or "", "distrito": fields.get("distrito"), "concelho": fields.get("concelho"), "freguesia": fields.get("freguesia"), "cp": fields.get("codigo_postal"), "localidade": fields.get("localidade"), "objeto": fields.get("objeto_social"), "capital": fields.get("capital_social"), "completa": bool(fields.get("has_constitution")), }, )