import hashlib import json as _json import secrets from datetime import date, datetime, timedelta, timezone from typing import Annotated, Any from fastapi import APIRouter, Body, Depends, Header, HTTPException, Path, status from pydantic import BaseModel from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.auth.deps import CurrentUser, require_admin from app.db import get_session from app.jobs.scheduler import ( list_scheduled_jobs, schedule_backfill_for_company, trigger_manual_scrape, ) from app.schemas.company import RaciusImportItem, RaciusImportResult from app.schemas.process import ScrapingLogOut from app.services.mj_parser import parse_mj_html from app.services.mj_people_parser import extract_people from app.services.nif import is_valid_nif router = APIRouter(prefix="/admin", tags=["admin"]) class ScrapeOptions(BaseModel): date_from: date | None = None date_to: date | None = None days: str | None = None # '15' | '30' | 'todos' types: list[str] | None = None # subset of ['internal','competitor','analysis'] # registo comercial: orçamento do backfill max_days: int | None = None max_items: int | None = None max_seconds: float | None = None @router.post("/scrape/{source}") async def scrape_now( source: str = Path(..., pattern="^(distribuicao|distribuicao_backfill|cire|ptdata|dre|contracts|registry_sweep|registry_expand|registry_content|registry_nacional)$"), _admin: Annotated[CurrentUser, Depends(require_admin)] = None, options: ScrapeOptions | None = Body(default=None), ) -> dict[str, str]: opts = options.model_dump(exclude_none=True) if options else {} task_id = await trigger_manual_scrape(source, opts) return {"source": source, "task_id": task_id} @router.get("/scheduled-jobs") async def scheduled_jobs( _admin: Annotated[CurrentUser, Depends(require_admin)], ) -> list[dict[str, object]]: """List all APScheduler jobs with their next run times.""" return list_scheduled_jobs() @router.get("/scraping-logs", response_model=list[ScrapingLogOut]) async def scraping_logs( session: Annotated[AsyncSession, Depends(get_session)], _admin: Annotated[CurrentUser, Depends(require_admin)], limit: int = 100, ) -> list[ScrapingLogOut]: rows = ( await session.execute( text( """ SELECT id::text, source, started_at, finished_at, status, rows_seen, rows_new, error, params, progress_current, progress_total FROM scraping_logs ORDER BY started_at DESC LIMIT :lim """ ), {"lim": limit}, ) ).all() return [ ScrapingLogOut( id=r[0], source=r[1], started_at=r[2], finished_at=r[3], status=r[4], rows_seen=r[5] or 0, rows_new=r[6] or 0, error=r[7], params=r[8], progress_current=r[9] or 0, progress_total=r[10] or 0, ) for r in rows ] @router.post("/racius-import", response_model=RaciusImportResult) async def racius_import( session: Annotated[AsyncSession, Depends(get_session)], _admin: Annotated[CurrentUser, Depends(require_admin)], items: list[RaciusImportItem] = Body(...), ) -> RaciusImportResult: """Bulk-import closed/insolvent competitors from Racius. Each row lands as monitored=TRUE, active=FALSE (one-shot historical scrape only). After insert, the per-company backfill is scheduled — 180d distribuicao + all CIRE — so we capture the credor/executado footprint without enrolling the company in the daily cron (active=FALSE gates that).""" inserted: list[str] = [] skipped: list[dict[str, str]] = [] backfill_ids: list[str] = [] for item in items: if not is_valid_nif(item.nif): skipped.append({"nif": item.nif, "reason": "invalid NIF checksum"}) continue existing = ( await session.execute( text("SELECT id::text FROM companies WHERE nif = :n"), {"n": item.nif}, ) ).first() if existing: skipped.append({"nif": item.nif, "reason": "already exists"}) continue row = ( await session.execute( text( """ INSERT INTO companies (nif, legal_name, status, monitored, active, monitoring_type) VALUES (:nif, :legal, :status, TRUE, FALSE, :mt) RETURNING id::text """ ), { "nif": item.nif, "legal": item.legal_name, "status": item.status, "mt": item.monitoring_type, }, ) ).first() await session.commit() inserted.append(item.nif) schedule_backfill_for_company(row[0], item.nif) backfill_ids.append(row[0]) return RaciusImportResult( inserted=inserted, skipped=skipped, total_scheduled_backfills=len(backfill_ids), ) @router.get("/false-positive-reports") async def false_positive_reports( session: Annotated[AsyncSession, Depends(get_session)], _admin: Annotated[CurrentUser, Depends(require_admin)], limit: int = 200, ) -> list[dict[str, object]]: rows = ( await session.execute( text( """ SELECT id::text, company_id::text, company_legal_name, process_number, tribunal, source, parties_snapshot, flagged_by_email, flagged_at, reason FROM false_positive_reports ORDER BY flagged_at DESC LIMIT :lim """ ), {"lim": limit}, ) ).all() return [ { "id": r[0], "company_id": r[1], "company_legal_name": r[2], "process_number": r[3], "tribunal": r[4], "source": r[5], "parties_snapshot": r[6], "flagged_by_email": r[7], "flagged_at": r[8], "reason": r[9], } for r in rows ] # ─────────────────────────────────────────────────────────────────────── # MJ bookmarklet workflow — user solves captcha on publicacoes.mj.pt, # clicks our bookmarklet, which POSTs the page HTML to /admin/mj-import. # Short-lived (24h) tokens authenticate that call because the user's # session cookie isn't available cross-origin. # ─────────────────────────────────────────────────────────────────────── def _stream_extension_zip(root_path: str, zip_name: str): import io import zipfile from pathlib import Path from fastapi.responses import StreamingResponse root = Path(root_path) if not root.exists(): raise HTTPException( status.HTTP_404_NOT_FOUND, f"Extension source not found on server (expected {root_path}).", ) buf = io.BytesIO() with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z: for path in root.rglob("*"): if path.is_file(): z.write(path, arcname=str(path.relative_to(root))) buf.seek(0) return StreamingResponse( buf, media_type="application/zip", headers={"Content-Disposition": f'attachment; filename="{zip_name}"'}, ) @router.get("/mj-extension.zip") async def mj_extension_zip( _admin: Annotated[CurrentUser, Depends(require_admin)], ): """Stable per-NIF capturer ZIP. Served from the host-mounted /app/extension so file edits roll out without a backend rebuild.""" return _stream_extension_zip("/app/extension", "intel-grid-mj-capturer.zip") @router.get("/mj-extension-bulk.zip") async def mj_extension_bulk_zip( _admin: Annotated[CurrentUser, Depends(require_admin)], ): """Bulk (beta) extension ZIP — iterates every monitored NIF through one captcha-backed session. Separate from the stable extension so we can experiment without affecting the per-NIF flow.""" return _stream_extension_zip("/app/extension-bulk", "intel-grid-mj-bulk.zip") @router.get("/mj-monitored") async def mj_monitored( session: Annotated[AsyncSession, Depends(get_session)], x_intel_token: Annotated[str | None, Header(alias="X-Intel-Token")] = None, ) -> list[dict[str, Any]]: """Return every active internal+competitor NIF the Intel Grid wants captured, sorted by oldest-last-capture first so successive runs preferentially refresh the stalest entries.""" await _check_mj_token(session, x_intel_token) rows = ( await session.execute( text( """ SELECT c.id::text, c.nif, c.legal_name, c.monitoring_type, MAX(d.created_at) AS last_captured_at FROM companies c LEFT JOIN dre_publications d ON d.company_id = c.id AND d.source = 'mj' WHERE c.active = true AND c.monitoring_type IN ('internal', 'competitor') GROUP BY c.id, c.nif, c.legal_name, c.monitoring_type ORDER BY MAX(d.created_at) ASC NULLS FIRST, c.nif """ ), ) ).all() return [ { "id": r[0], "nif": r[1], "legal_name": r[2], "monitoring_type": r[3], "last_captured_at": r[4].isoformat() if r[4] else None, } for r in rows ] @router.post("/mj-token") async def mj_generate_token( session: Annotated[AsyncSession, Depends(get_session)], admin: Annotated[CurrentUser, Depends(require_admin)], ) -> dict[str, str]: """Issue a 24h token the admin embeds in their bookmarklet. Each call generates a new token and supersedes any previous one for the same user — simpler than tracking devices. Old tokens still work until they expire naturally.""" token = "mj_" + secrets.token_urlsafe(32) expires = datetime.now(tz=timezone.utc) + timedelta(hours=24) await session.execute( text( """ INSERT INTO mj_import_tokens (token, user_id, expires_at) VALUES (:tok, :uid, :exp) """ ), {"tok": token, "uid": admin.id, "exp": expires}, ) await session.commit() return { "token": token, "expires_at": expires.isoformat(), } async def _check_mj_token( session: AsyncSession, x_intel_token: str | None, ) -> str: if not x_intel_token: raise HTTPException(status.HTTP_401_UNAUTHORIZED, "missing X-Intel-Token") row = ( await session.execute( text( """ SELECT user_id::text FROM mj_import_tokens WHERE token = :t AND expires_at > now() """ ), {"t": x_intel_token}, ) ).first() if not row: raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid/expired token") return row[0] @router.get("/mj-existing") async def mj_existing( nif: str, session: Annotated[AsyncSession, Depends(get_session)], x_intel_token: Annotated[str | None, Header(alias="X-Intel-Token")] = None, ) -> dict[str, Any]: """Return every MJ publication already captured for this NIF so the extension can skip re-fetching the detail of rows we already have. `has_detail` is True when raw_json.detail_body is non-empty — those rows should be skipped entirely on the next capture run to avoid hammering publicacoes.mj.pt.""" await _check_mj_token(session, x_intel_token) co = ( await session.execute( text("SELECT id::text FROM companies WHERE nif = :n"), {"n": nif}, ) ).first() if not co: return {"company_id": None, "rows": []} rows = ( await session.execute( text( """ SELECT to_char(date, 'YYYY-MM-DD') AS d, title, COALESCE(length(raw_json->>'detail_body'), 0) > 0 AS has_detail FROM dre_publications WHERE company_id = :c AND source = 'mj' """ ), {"c": co[0]}, ) ).all() return { "company_id": co[0], "rows": [ {"date": r.d, "title": r.title, "has_detail": bool(r.has_detail)} for r in rows ], } class MjImportRow(BaseModel): date: str | None = None nif: str | None = None entity: str | None = None concelho: str | None = None acto: str | None = None detail_html: str | None = None class MjImportPayload(BaseModel): # Legacy bookmarklet shape — posts the whole list-page HTML for summary- # only capture. Kept for backwards compat while the extension rolls out. html: str | None = None # New extension shape — per-row dicts with optional detail HTML. rows: list[MjImportRow] | None = None # Either payload form can pass these. nif: str | None = None # treated as nif_hint when rows[] is used nif_hint: str | None = None company_id: str | None = None # optional: forces target company @router.post("/mj-import") async def mj_import( payload: Annotated[MjImportPayload, Body()], session: Annotated[AsyncSession, Depends(get_session)], x_intel_token: Annotated[str | None, Header(alias="X-Intel-Token")] = None, ) -> dict[str, Any]: """Receive HTML scraped by the admin's browser bookmarklet, parse publication rows, insert into dre_publications with source='mj'. Auth via X-Intel-Token (short-lived, issued by /admin/mj-token). No session cookie because the POST comes from publicacoes.mj.pt origin (cross-origin, no cookies sent per our CORS config).""" if not x_intel_token: raise HTTPException(status.HTTP_401_UNAUTHORIZED, "missing X-Intel-Token") tok_row = ( await session.execute( text( """ SELECT user_id FROM mj_import_tokens WHERE token = :t AND expires_at > now() """ ), {"t": x_intel_token}, ) ).first() if not tok_row: raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid/expired token") # Bump usage counter (ignore errors, best-effort) await session.execute( text( """ UPDATE mj_import_tokens SET last_used_at = now(), uses = uses + 1 WHERE token = :t """ ), {"t": x_intel_token}, ) await session.commit() # Parse — supports two shapes. Extension sends rows[] with optional # detail_html. Legacy bookmarklet sends a single html blob. from app.services.mj_parser import parse_detail_html nipc_hint = payload.nif or payload.nif_hint items: list[dict[str, Any]] = [] if payload.rows: for r in payload.rows: d = r.date parsed_date = None if d: from datetime import datetime as _dt for fmt in ("%Y-%m-%d", "%d-%m-%Y", "%d/%m/%Y"): try: parsed_date = _dt.strptime(d, fmt).date() break except ValueError: continue detail = parse_detail_html(r.detail_html) if r.detail_html else {} # Summary stays short — entity · concelho · acto. The full detail # body is stored in raw_json.detail_body and rendered in the UI # modal, not inlined into the list view. summary_parts = [p for p in (r.entity, r.concelho, r.acto) if p] items.append({ "title": (r.acto or "")[:500], "summary": " · ".join(summary_parts)[:2000], "date": parsed_date, "type": r.acto, "source_url": None, "source": "mj", "nipc": r.nif, "raw": { "date": r.date, "nif": r.nif, "entity": r.entity, "concelho": r.concelho, "acto": r.acto, "detail_fields": detail.get("fields"), "detail_body": detail.get("body"), }, }) elif payload.html: items = parse_mj_html(payload.html, nipc_hint=nipc_hint) if not items: return {"inserted": 0, "skipped": 0, "parsed": 0, "message": "No publication rows found."} # Resolve target company. Precedence: explicit company_id → nif hint → row NIPC. target_cid: str | None = payload.company_id if not target_cid: target_nif = nipc_hint if not target_nif: # infer from the first row for it in items: if it.get("nipc"): target_nif = it["nipc"] break if target_nif: row = ( await session.execute( text("SELECT id::text FROM companies WHERE nif = :n"), {"n": target_nif}, ) ).first() if row: target_cid = row[0] if not target_cid: return {"inserted": 0, "skipped": len(items), "parsed": len(items), "message": "Could not resolve target company. Pass ?nif= or ?company_id."} # Upsert via dre_publications with source='mj' (same shape as DRE). # When the extension re-sends a row that was already captured (e.g. the # bookmarklet saw it first, without detail), and this time has detail_html # parsed into raw.detail_body, enrich the existing record. Only overwrite # when the incoming payload carries a non-empty detail_body, so a later # bare-row import can't wipe previously captured detail. from app.services.dre_classifier import classify inserted = 0 enriched = 0 for it in items: d = it.get("date") key = f"{target_cid}|mj|{(it.get('title') or '').strip().lower()}|{d or ''}" h = hashlib.sha256(key.encode("utf-8")).hexdigest() cls = classify(it.get("title"), it.get("summary")) has_detail = bool((it.get("raw") or {}).get("detail_body")) res = await session.execute( text( """ INSERT INTO dre_publications (company_id, title, summary, date, type, source_url, raw_json, dedup_hash, relevance, change_kind, source) VALUES (:cid, :t, :s, :d, :ty, :u, CAST(:r AS JSONB), :h, :rel, :ck, 'mj') ON CONFLICT (dedup_hash) DO UPDATE SET summary = CASE WHEN :has_detail THEN EXCLUDED.summary ELSE dre_publications.summary END, raw_json = CASE WHEN :has_detail THEN EXCLUDED.raw_json ELSE dre_publications.raw_json END, relevance = CASE WHEN :has_detail THEN EXCLUDED.relevance ELSE dre_publications.relevance END, change_kind= CASE WHEN :has_detail THEN EXCLUDED.change_kind ELSE dre_publications.change_kind END RETURNING (xmax = 0) AS is_insert """ ), { "cid": target_cid, "t": it.get("title") or "", "s": it.get("summary"), "d": d, "ty": it.get("type"), "u": it.get("source_url"), "r": _json.dumps(it.get("raw") or {}, default=str), "h": h, "rel": cls.relevance, "ck": cls.change_kind, "has_detail": has_detail, }, ) row = res.first() pub_id = None if row: if row[0]: inserted += 1 elif has_detail: enriched += 1 # RETURNING still returns the row id regardless of insert vs # update, but our query uses (xmax=0) so we need a second # fetch for the id. Do it inline to keep the loop simple. # Extract people from detail_body and persist. We need the id of # the publication we just upserted, so look it up by dedup_hash. if has_detail: pub_row = ( await session.execute( text("SELECT id::text FROM dre_publications WHERE dedup_hash = :h"), {"h": h}, ) ).first() if pub_row: pub_id = pub_row[0] detail_body = (it.get("raw") or {}).get("detail_body") or "" for p in extract_people(detail_body): await session.execute( text( """ INSERT INTO publication_people (publication_id, company_id, person_nif, person_name, cargo, event, act_date) VALUES (:pid, :cid, :nif, :nm, :cg, :ev, :dt) ON CONFLICT (publication_id, person_nif) DO UPDATE SET person_name = EXCLUDED.person_name, cargo = EXCLUDED.cargo, event = EXCLUDED.event, act_date = EXCLUDED.act_date """ ), { "pid": pub_id, "cid": target_cid, "nif": p["nif"], "nm": p.get("name"), "cg": p.get("cargo"), "ev": p.get("event") or "other", "dt": d, }, ) # When the person is a legal entity (NIPC starts with # 5), auto-register them as monitoring_type='related' # if we don't know them yet. This makes the sócio # clickable in the UI and discoverable via the # "Relacionadas" tab in the Concorrentes section. if p["nif"].startswith("5") and p.get("name"): await session.execute( text( """ INSERT INTO companies (nif, legal_name, monitoring_type, monitored, active, status) VALUES (:nif, :nm, 'related', false, true, 'active') ON CONFLICT (nif) DO NOTHING """ ), {"nif": p["nif"], "nm": p["name"][:200]}, ) await session.commit() return { "inserted": inserted, "enriched": enriched, "skipped": len(items) - inserted - enriched, "parsed": len(items), "company_id": target_cid, "message": f"Imported {inserted} new, enriched {enriched} with detail.", }