0pt/services/api/sources/connectors.py

"""
0pt data engine — COMPLIANT source connectors.

Legal sourcing only (CLAUDE.md §1, docs/06). Priority per field/market:
  1. official API / bulk   2. authorized public web (robots + rate limits)
  3. licensed partner data  4. otherwise -> level 0 "ej tillgänglig / licens".

EXPLICITLY NOT PRESENT (and must never be added here):
  WAF/bot bypass, FlareSolverr, stealth/undetected browsers, proxy/fingerprint
  rotation to evade bans, scraping behind login walls, fake-account creation.

Every connector returns a normalized company dict + per-field provenance rows
so nothing ships unsourced. `Provenance` (below) carries the full field set
BEFORE_STATE.md §3 found missing from the actual dataflow — this repo's
5-field bridge_provenance was: source, confidence, level, and a DATE-only
freshness; the concept "bridge_provenance" as a named thing did not exist
anywhere in the repo at all. This adds, per field: retrieval TIMESTAMP
(distinct from the legacy date-only `freshness`), the transform applied,
connector/schema version, the producing system, and two slots
(`quality`/`entity_resolution`) that downstream stages (quality bridge,
resolve.py) fill in as the record moves through the pipeline — they start
`None` here and are never backfilled with a guess.
Terms change — VERIFY each source's current access/licence before relying.
"""
from __future__ import annotations

import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any

import httpx

# A polite client that identifies itself and obeys timeouts/rate limits.
USER_AGENT = "0pt-data-engine/0.1 (+https://0pt.app; SIAX Technology AB; hej@siax.io)"
_TIMEOUT = httpx.Timeout(15.0, connect=10.0)

# Bumped when a connector's parsing/field-mapping logic changes in a way that
# could shift output shape — recorded per-field as `Provenance.connector_version`
# so a downstream consumer can tell which parsing logic produced a value.
CONNECTOR_VERSIONS: dict[str, str] = {
    "gleif": "gleif-connector@1",
    "brreg": "brreg-connector@1",
    "companies_house": "companies-house-connector@1",
    "prh": "prh-connector@1",
    "cvr": "cvr-connector@1",
    "rdw": "rdw-connector@1",
}

PRODUCER = "0pt-engine"

# P014: statuses a compliant source uses to signal "stop asking me this way"
# (rate-limited, forbidden, or a legal/geo block) rather than "this record
# doesn't exist". Collapsing these into an ordinary None (as every connector
# did before this pass) is not itself a compliance violation -- nothing here
# ever escalates to a bypass -- but it silently discarded a real signal:
# adaptive_router.py had no way to know a source had just told it to back
# off, so nothing could avoid hammering a rate-limited endpoint on the very
# next record. See SourceBlockedError and adaptive_router.record_blocked.
BLOCKED_STATUSES = frozenset({401, 403, 429, 451})


class SourceBlockedError(Exception):
    """
    Raised by a connector when the source responds with one of
    BLOCKED_STATUSES. This is informational, never an instruction to retry
    harder or escalate -- callers (pipeline.py's _fetch loop) catch it only
    to record the block via adaptive_router.record_blocked and move on to
    the next compliant source in the plan, exactly as they already would for
    a plain None. Compliance stance (CLAUDE.md §1) is unchanged: a block is
    always a stop signal for THAT source, never a cue to bypass it.
    """

    def __init__(self, source_id: str, status_code: int):
        self.source_id = source_id
        self.status_code = status_code
        super().__init__(f"{source_id} signaled a block (HTTP {status_code})")


def _raise_if_blocked(source_id: str, response: httpx.Response) -> None:
    if response.status_code in BLOCKED_STATUSES:
        raise SourceBlockedError(source_id, response.status_code)


def _client() -> httpx.AsyncClient:
    return httpx.AsyncClient(
        headers={"User-Agent": USER_AGENT, "Accept": "application/json"},
        timeout=_TIMEOUT,
        follow_redirects=True,
    )


@dataclass
class Provenance:
    field_name: str
    source_id: str
    confidence: float
    level: int          # 0 unavailable/licensed, 1 single, 2 partial, 3 corroborated
    freshness: str | None = None       # legacy: retrieval DATE only (kept for back-compat)
    retrieved_at: str | None = None    # full ISO-8601 UTC timestamp of the actual fetch
    transform: str | None = None       # e.g. "raw_fetch", "normalized", "entity_resolution_merge"
    connector_version: str | None = None  # e.g. "brreg-connector@1" — see CONNECTOR_VERSIONS
    producer: str = PRODUCER           # which system produced this field's value
    quality: dict[str, Any] | None = None          # filled by the quality bridge, if/when it runs (never guessed)
    entity_resolution: dict[str, Any] | None = None  # filled by resolve.py on merge, never at fetch time
    artifact_ref: str | None = None    # delivery artifact reference (ST0RE), set at delivery time — not yet wired


def make_provenance(
    field_name: str,
    source_id: str,
    confidence: float,
    level: int,
    *,
    transform: str = "raw_fetch",
) -> Provenance:
    """Factory used by every connector below so the new provenance fields
    (retrieved_at, connector_version, producer) are populated consistently
    instead of each call site re-deriving them ad hoc."""
    now = datetime.now(timezone.utc)
    return Provenance(
        field_name=field_name,
        source_id=source_id,
        confidence=confidence,
        level=level,
        freshness=now.date().isoformat(),
        retrieved_at=now.isoformat(),
        transform=transform,
        connector_version=CONNECTOR_VERSIONS.get(source_id),
        producer=PRODUCER,
    )


@dataclass
class CompanyResult:
    org_nr: str | None = None
    market: str | None = None
    lei: str | None = None
    name: str | None = None
    status: str | None = None
    address: dict[str, Any] | None = None
    website: str | None = None
    provenance: list[Provenance] = field(default_factory=list)
    source: str = ""


# ── GLEIF / LEI (global cross-border key) — open API ────────────────────────
async def gleif_by_lei(lei: str) -> CompanyResult | None:
    url = f"https://api.gleif.org/api/v1/lei-records/{lei}"
    async with _client() as c:
        r = await c.get(url, headers={"Accept": "application/vnd.api+json"})
        _raise_if_blocked("gleif", r)
        if r.status_code != 200:
            return None
        a = (r.json().get("data") or {}).get("attributes") or {}
        entity = a.get("entity") or {}
        legal = entity.get("legalAddress") or {}
        res = CompanyResult(
            lei=lei,
            name=(entity.get("legalName") or {}).get("name"),
            status=entity.get("status"),
            address={"lines": legal.get("addressLines"), "city": legal.get("city"),
                     "country": legal.get("country"), "postal": legal.get("postalCode")},
            source="gleif",
        )
        res.provenance += [
            make_provenance("lei", "gleif", 0.99, 3),
            make_provenance("name", "gleif", 0.97, 3),
        ]
        return res


# ── Brønnøysundregistrene (NO) — open API (NLOD). VERIFY terms. ─────────────
async def brreg_by_orgnr(orgnr: str) -> CompanyResult | None:
    url = f"https://data.brreg.no/enhetsregisteret/api/enheter/{orgnr}"
    async with _client() as c:
        r = await c.get(url)
        _raise_if_blocked("brreg", r)
        if r.status_code != 200:
            return None
        d = r.json()
        res = CompanyResult(
            org_nr=orgnr, market="NO",
            name=d.get("navn"),
            status="dissolved" if d.get("konkurs") else "active",
            address={"raw": d.get("forretningsadresse")},
            source="brreg",
        )
        res.provenance += [
            make_provenance("name", "brreg", 0.97, 3),
            make_provenance("status", "brreg", 0.95, 3),
            make_provenance("address", "brreg", 0.95, 3),
        ]
        return res


# ── Companies House (UK) — free API, key required. VERIFY terms. ────────────
async def companies_house_by_number(number: str) -> CompanyResult | None:
    key = os.getenv("COMPANIES_HOUSE_API_KEY")
    if not key:
        # No key -> deliver honestly as unavailable, never fabricate.
        return CompanyResult(org_nr=number, market="GB", source="companies_house",
                             provenance=[make_provenance("name", "companies_house", 0.0, 0, transform="unavailable_no_key")])
    url = f"https://api.company-information.service.gov.uk/company/{number}"
    async with _client() as c:
        r = await c.get(url, auth=(key, ""))
        _raise_if_blocked("companies_house", r)
        if r.status_code != 200:
            return None
        d = r.json()
        res = CompanyResult(
            org_nr=number, market="GB",
            name=d.get("company_name"),
            status=d.get("company_status"),
            address={"raw": d.get("registered_office_address")},
            source="companies_house",
        )
        res.provenance += [
            make_provenance("name", "companies_house", 0.97, 3),
            make_provenance("status", "companies_house", 0.95, 3),
        ]
        return res


# ── PRH avoindata (FI) — open API (CC-BY). VERIFY terms. ────────────────────
async def prh_by_businessid(business_id: str) -> CompanyResult | None:
    url = f"https://avoindata.prh.fi/opendata-ytj-api/v3/companies?businessId={business_id}"
    async with _client() as c:
        r = await c.get(url)
        _raise_if_blocked("prh", r)
        if r.status_code != 200:
            return None
        items = r.json().get("companies") or r.json().get("results") or []
        if not items:
            return None
        d = items[0]
        names = d.get("names") or []
        name = names[0].get("name") if names else d.get("name")
        res = CompanyResult(org_nr=business_id, market="FI", name=name, status=d.get("status"), source="prh")
        if name:
            res.provenance.append(make_provenance("name", "prh", 0.95, 3))
        return res


# ── CVR (DK) via cvrapi.dk — free for low volume (descriptive UA). VERIFY. ───
async def cvr_by_number(cvr: str) -> CompanyResult | None:
    url = f"https://cvrapi.dk/api?search={cvr}&country=dk"
    async with _client() as c:
        r = await c.get(url)
        _raise_if_blocked("cvr", r)
        if r.status_code != 200:
            return None
        d = r.json()
        if not d.get("name"):
            return None
        res = CompanyResult(
            org_nr=cvr, market="DK", name=d.get("name"),
            status="ceased" if d.get("enddate") else "active",
            address={"raw": d.get("address")}, source="cvr",
        )
        res.provenance += [make_provenance("name", "cvr", 0.93, 3), make_provenance("address", "cvr", 0.9, 2)]
        return res


# ── RDW (NL) open vehicle data — Socrata. Technical/fleet only, no owner PII. ─
async def rdw_vehicle(kenteken: str) -> dict[str, Any] | None:
    plate = kenteken.upper().replace("-", "").replace(" ", "")
    url = f"https://opendata.rdw.nl/resource/m9d7-ebf2.json?kenteken={plate}"
    async with _client() as c:
        r = await c.get(url)
        r.raise_for_status()
        arr = r.json()
        if not isinstance(arr, list) or any(not isinstance(item, dict) for item in arr):
            raise ValueError("invalid RDW response")
        if not arr:
            return None
        d = arr[0]
        p = make_provenance("make", "rdw", 0.9, 3)
        return {
            "reg_plate": kenteken, "market": "NL",
            "make": d.get("merk"), "model": d.get("handelsbenaming"),
            "year": (d.get("datum_eerste_toelating") or "")[:4] or None,
            "source": "rdw",
            # Dict shape (not the Provenance dataclass) — this endpoint predates it
            # and is returned directly as an API response body, not merged through
            # resolve.py. Field names kept aligned with Provenance's for consistency.
            "provenance": [{
                "field": p.field_name, "source_id": p.source_id, "confidence": p.confidence, "level": p.level,
                "freshness": p.freshness, "retrieved_at": p.retrieved_at, "transform": p.transform,
                "connector_version": p.connector_version, "producer": p.producer,
            }],
        }


# ── OpenSanctions screening — search API. VERIFY commercial terms / key. ─────
async def opensanctions_screen(name: str) -> dict[str, Any]:
    url = "https://api.opensanctions.org/search/default"
    async with _client() as c:
        r = await c.get(url, params={"q": name, "limit": 5})
        r.raise_for_status()
        payload = r.json()
        if not isinstance(payload, dict) or not isinstance(payload.get("results"), list) or any(
            not isinstance(item, dict) for item in payload["results"]
        ):
            raise ValueError("invalid OpenSanctions response")
        results = payload["results"]
        return {
            "source": "opensanctions",
            "matches": [{"name": x.get("caption"), "score": x.get("score"), "schema": x.get("schema")} for x in results],
        }


CONNECTORS = {
    "gleif": gleif_by_lei,
    "brreg": brreg_by_orgnr,
    "companies_house": companies_house_by_number,
    "prh": prh_by_businessid,
    "cvr": cvr_by_number,
}