adminandClaude Opus 4.8
be6909bce2
Edge /api/vehicles (RDW NL, live now) + /api/screen (OpenSanctions, live now), both defer to engine when ENGINE_URL set. Engine: /vehicles/{plate} + /screen endpoints; workers/resolve (blocking+fuzzy entity resolution, now used by pipeline); workers/queue (in-process job queue); sources/financials (Companies House accounts + XBRL path). py_compile + tsc + next build green.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
"""
|
|
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
|
|
seeds: dict[str, list[str]] = config.get("seeds") or 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,
|
|
}
|