"""Persistência dos anúncios CIRE varridos nacionalmente. Guarda o anúncio, as suas partes, e — quando o acto abre prazo de reclamação — os campos extraídos do PDF. O cruzamento com a watchlist vive em `insolvency_matching.py`, chamado a seguir a cada lote. """ import json import logging from typing import Any from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.services.cire_announcement_parser import ( compute_deadline, parse_announcement_text, ) from app.services.dedup import announcement_hash logger = logging.getLogger(__name__) async def reparse_stored_details(session: AsyncSession, only_missing: bool = True) -> int: """Reaplica o parser ao `pdf_text` já guardado, sem voltar ao CITIUS. Os anúncios variam de tribunal para tribunal e o parser vai ganhando padrões com o tempo; sem isto, cada melhoria só valeria para o que fosse varrido a partir daí, e o histórico ficava com prazos por extrair para sempre. """ rows = ( await session.execute( text( f""" SELECT id::text, pdf_text, pub_date, opens_claim_window FROM cire_announcements WHERE pdf_text IS NOT NULL {"AND claim_days IS NULL" if only_missing else ""} """ ) ) ).all() updated = 0 for ann_id, pdf_text, pub_date, opens in rows: detail = parse_announcement_text(pdf_text) if detail.get("claim_days") is None and not any( detail.get(k) for k in ("debtor_nif", "admin_nif", "insolvency_date") ): continue deadline, estimated = compute_deadline( pub_date, detail.get("claim_days"), detail.get("editos_days"), opens ) await session.execute( text( """ UPDATE cire_announcements SET claim_days = COALESCE(:claim_days, claim_days), claim_days_source = COALESCE(:src, claim_days_source), editos_days = COALESCE(:editos, editos_days), claim_deadline = COALESCE(:deadline, claim_deadline), claim_deadline_estimated = CASE WHEN :claim_days IS NOT NULL THEN :estimated ELSE claim_deadline_estimated END, insolvency_date = COALESCE(:idate, insolvency_date), debtor_nif = COALESCE(:dnif, debtor_nif), debtor_name = COALESCE(:dname, debtor_name), debtor_address = COALESCE(:daddr, debtor_address), admin_name = COALESCE(:aname, admin_name), admin_email = COALESCE(:aemail, admin_email), admin_nif = COALESCE(:anif, admin_nif) WHERE id = :id """ ), { "id": ann_id, "claim_days": detail.get("claim_days"), "src": detail.get("claim_days_source"), "editos": detail.get("editos_days"), "deadline": deadline, "estimated": estimated, "idate": detail.get("insolvency_date"), "dnif": detail.get("debtor_nif"), "dname": detail.get("debtor_name"), "daddr": detail.get("debtor_address"), "aname": detail.get("admin_name"), "aemail": detail.get("admin_email"), "anif": detail.get("admin_nif"), }, ) updated += 1 # os prazos podem ter mudado, portanto os hits têm de acompanhar await session.execute( text( """ UPDATE insolvency_hits h SET claim_deadline = a.claim_deadline FROM cire_announcements a WHERE a.id = h.announcement_id AND h.claim_deadline IS DISTINCT FROM a.claim_deadline """ ) ) await session.commit() logger.info("reparse de anúncios: %d actualizados de %d candidatos", updated, len(rows)) return updated async def upsert_announcement( session: AsyncSession, row: dict[str, Any] ) -> tuple[str, bool]: """Insere ou actualiza um anúncio. Devolve (id, is_new). Ao reencontrar um anúncio já conhecido só se toca no `last_seen_at` e nos campos do PDF — que podem chegar numa corrida posterior, se o detalhe não foi puxado à primeira. """ h = announcement_hash( row.get("referencia"), row.get("process_number"), row.get("ato"), row.get("pub_date") ) detail = row.get("detail") or {} opens = bool(row.get("wants_pdf")) deadline, estimated = compute_deadline( row.get("pub_date"), detail.get("claim_days"), detail.get("editos_days"), opens ) result = await session.execute( text( """ INSERT INTO cire_announcements ( referencia, tribunal, juizo, ato, ato_group, process_number, species, pub_date, action_date, raw, pdf_token, has_pdf, pdf_text, detail_fetched_at, claim_days, claim_days_source, editos_days, claim_deadline, claim_deadline_estimated, opens_claim_window, insolvency_date, debtor_nif, debtor_name, debtor_address, admin_name, admin_email, admin_nif, dedup_hash ) VALUES ( :referencia, :tribunal, :juizo, :ato, :ato_group, :process_number, :species, :pub_date, :action_date, CAST(:raw AS JSONB), :pdf_token, :has_pdf, :pdf_text, CASE WHEN :has_pdf THEN now() ELSE NULL END, :claim_days, :claim_days_source, :editos_days, :claim_deadline, :claim_deadline_estimated, :opens_claim_window, :insolvency_date, :debtor_nif, :debtor_name, :debtor_address, :admin_name, :admin_email, :admin_nif, :dedup_hash ) ON CONFLICT (dedup_hash) DO UPDATE SET last_seen_at = now(), -- só enriquecemos: nunca apagamos um detalhe já obtido com um NULL pdf_text = COALESCE(EXCLUDED.pdf_text, cire_announcements.pdf_text), has_pdf = cire_announcements.has_pdf OR EXCLUDED.has_pdf, detail_fetched_at = COALESCE( EXCLUDED.detail_fetched_at, cire_announcements.detail_fetched_at), claim_days = COALESCE(EXCLUDED.claim_days, cire_announcements.claim_days), claim_days_source = COALESCE( EXCLUDED.claim_days_source, cire_announcements.claim_days_source), editos_days = COALESCE(EXCLUDED.editos_days, cire_announcements.editos_days), claim_deadline = COALESCE( EXCLUDED.claim_deadline, cire_announcements.claim_deadline), claim_deadline_estimated = CASE WHEN EXCLUDED.claim_days IS NOT NULL THEN EXCLUDED.claim_deadline_estimated ELSE cire_announcements.claim_deadline_estimated END, insolvency_date = COALESCE( EXCLUDED.insolvency_date, cire_announcements.insolvency_date), debtor_nif = COALESCE(EXCLUDED.debtor_nif, cire_announcements.debtor_nif), debtor_name = COALESCE(EXCLUDED.debtor_name, cire_announcements.debtor_name), debtor_address = COALESCE( EXCLUDED.debtor_address, cire_announcements.debtor_address), admin_name = COALESCE(EXCLUDED.admin_name, cire_announcements.admin_name), admin_email = COALESCE(EXCLUDED.admin_email, cire_announcements.admin_email), admin_nif = COALESCE(EXCLUDED.admin_nif, cire_announcements.admin_nif) RETURNING id::text, (xmax = 0) AS is_new """ ), { "referencia": row.get("referencia"), "tribunal": row.get("tribunal"), "juizo": row.get("juizo"), "ato": row.get("ato"), "ato_group": row.get("ato_group"), "process_number": row.get("process_number"), "species": row.get("species"), "pub_date": row.get("pub_date"), "action_date": row.get("action_date"), "raw": json.dumps(row.get("raw") or {}, ensure_ascii=False, default=str), "pdf_token": row.get("pdf_token"), "has_pdf": bool(row.get("pdf_text")), "pdf_text": row.get("pdf_text"), "claim_days": detail.get("claim_days"), "claim_days_source": detail.get("claim_days_source"), "editos_days": detail.get("editos_days"), "claim_deadline": deadline, "claim_deadline_estimated": estimated, "opens_claim_window": opens, "insolvency_date": detail.get("insolvency_date"), "debtor_nif": detail.get("debtor_nif"), "debtor_name": detail.get("debtor_name"), "debtor_address": detail.get("debtor_address"), "admin_name": detail.get("admin_name"), "admin_email": detail.get("admin_email"), "admin_nif": detail.get("admin_nif"), "dedup_hash": h, }, ) ann_id, is_new = result.first() if is_new: for p in row.get("parties") or []: nif = (p.get("nif") or "").strip() or None if nif and (not nif.isdigit() or len(nif) != 9): nif = None await session.execute( text( """ INSERT INTO cire_announcement_parties (announcement_id, role, name, nif) VALUES (:aid, :role, :name, :nif) """ ), {"aid": ann_id, "role": p.get("role"), "name": p.get("name"), "nif": nif}, ) return str(ann_id), bool(is_new)