adminandClaude Opus 4.8
31128a71c8
FastAPI orchestrator (health/sources/estimates/orders/jobs per docs/04), adaptive router (compliant methods only, no bypass tiers), pipeline (sourcing->extraction->structuring/dedup->fusion->QA with per-field provenance, drops < confidence target), live connectors GLEIF/Brreg/Companies House (PRH/CVR/Bolagsverket gated to level-0 until terms verified). sources.yaml registry. infra/docker-compose + Caddyfile for the VPS. FlareSolverr/stealth/proxy/fake-account paths intentionally NOT carried over (CLAUDE.md §1). py_compile clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
90 lines
3.3 KiB
Python
90 lines
3.3 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 rapidfuzz import fuzz
|
|
|
|
from sources.connectors import (
|
|
CompanyResult,
|
|
brreg_by_orgnr,
|
|
companies_house_by_number,
|
|
gleif_by_lei,
|
|
)
|
|
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)
|
|
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 — dedup by fuzzy name within market
|
|
deduped: list[dict[str, Any]] = []
|
|
for r in records:
|
|
dup = next((d for d in deduped if d["market"] == r["market"] and fuzz.ratio(d["name"], r["name"]) > 92), None)
|
|
if dup:
|
|
dup["confidence"] = min(0.99, dup["confidence"] + 0.1) # corroboration bump
|
|
else:
|
|
deduped.append(r)
|
|
|
|
# 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,
|
|
}
|