0pt/services/api/workers/pipeline.py

"""
Production pipeline (docs/06): sourcing -> extraction -> structuring ->
enrichment/fusion -> QA. Compliant connectors only; every field gets provenance;
rows below the order's confidence target are dropped (deliver fewer, never fabricate).
"""
from __future__ import annotations

from typing import Any

from workers.resolve import resolve

from sources.connectors import (
    CompanyResult,
    brreg_by_orgnr,
    companies_house_by_number,
    cvr_by_number,
    gleif_by_lei,
    prh_by_businessid,
)
from workers.adaptive_router import plan_for_market

# Minimal built-in seed so a job is demonstrable without customer input.
DEMO_SEEDS: dict[str, list[str]] = {
    "NO": ["923609016"],            # brreg orgnr (example)
    "GB": ["00000006"],             # companies house number (example)
}


async def _fetch(source_id: str, key: str) -> CompanyResult | None:
    if source_id == "gleif":
        return await gleif_by_lei(key)
    if source_id == "brreg":
        return await brreg_by_orgnr(key)
    if source_id == "companies_house":
        return await companies_house_by_number(key)
    if source_id == "prh":
        return await prh_by_businessid(key)
    if source_id == "cvr":
        return await cvr_by_number(key)
    return None


async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]:
    markets: list[str] = config.get("markets", []) or ["NO"]
    threshold = float(config.get("confidence", 92)) / 100.0
    # Use caller-supplied seeds when present (an explicit empty {} is honoured so a
    # real job is never silently backfilled with DEMO_SEEDS), but treat an explicit
    # null the same as "missing" -- `config.get("seeds") is not None` avoids a 500
    # (None.get) on client-controlled `{"seeds": null}`, which the old `... or ...`
    # idiom tolerated and the `"seeds" in config` form regressed.
    seeds: dict[str, list[str]] = config["seeds"] if config.get("seeds") is not None else DEMO_SEEDS

    records: list[dict[str, Any]] = []
    per_market: dict[str, int] = {}

    # 1) sourcing + 2) extraction
    for market in markets:
        keys = seeds.get(market, [])
        plan = plan_for_market(market)
        got = 0
        for key in keys:
            for source_id in plan:
                try:
                    res = await _fetch(source_id, key)
                except Exception:
                    res = None
                if res and res.name:
                    records.append({
                        "market": market, "name": res.name, "status": res.status,
                        "lei": res.lei, "org_nr": res.org_nr, "source": res.source,
                        "provenance": [p.__dict__ for p in res.provenance],
                        "confidence": max((p.confidence for p in res.provenance), default=0.0),
                    })
                    got += 1
                    break  # first compliant source that answers wins
        per_market[market] = got

    # 3) structuring — entity resolution (blocking + fuzzy match + corroboration)
    deduped = resolve(records)

    # 4) enrichment/fusion already folded into confidence above.
    # 5) QA — drop rows below the order's confidence threshold.
    delivered = [r for r in deduped if r["confidence"] >= threshold]
    dropped = len(deduped) - len(delivered)
    avg_conf = round(sum(r["confidence"] for r in delivered) / len(delivered), 3) if delivered else 0.0

    return {
        "stage": "qa", "pct": 100, "state": "done",
        "metrics": {
            "rows": len(delivered), "dropped_below_threshold": dropped,
            "per_market": per_market, "avg_confidence": avg_conf,
        },
        "records": delivered,
    }