152 lines
5.4 KiB
Python
152 lines
5.4 KiB
Python
"""Production pipeline: sourcing -> extraction -> canonical ER -> QA -> quality.
|
|
|
|
Identity decisions are owned by src/er/entity-resolution-service.ts. The
|
|
Python worker only applies canonical `match` decisions returned by its
|
|
internal bridge. If tenant context or the canonical bridge is unavailable,
|
|
records remain unmerged: false identity merges are never used as fallback.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from workers.resolve import resolve
|
|
from workers.quality_bridge import run_quality_bridge
|
|
from sources.connectors import (
|
|
CompanyResult,
|
|
SourceBlockedError,
|
|
brreg_by_orgnr,
|
|
companies_house_by_number,
|
|
cvr_by_number,
|
|
gleif_by_lei,
|
|
prh_by_businessid,
|
|
)
|
|
from workers.adaptive_router import default_router, plan_for_market
|
|
|
|
logger = logging.getLogger(__name__)
|
|
PIPELINE_VERSION = "0pt-pipeline@2"
|
|
|
|
DEMO_SEEDS: dict[str, list[str]] = {
|
|
"NO": ["923609016"],
|
|
"GB": ["00000006"],
|
|
}
|
|
|
|
|
|
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["seeds"] if config.get("seeds") is not None else DEMO_SEEDS
|
|
tenant_id = config.get("tenant_id")
|
|
project_id = config.get("project_id", "p_default")
|
|
correlation_id = config.get("correlation_id")
|
|
|
|
records: list[dict[str, Any]] = []
|
|
per_market: dict[str, int] = {}
|
|
|
|
for market in markets:
|
|
keys = seeds.get(market, [])
|
|
got = 0
|
|
for key in keys:
|
|
plan = plan_for_market(market)
|
|
for source_id in plan:
|
|
try:
|
|
res = await _fetch(source_id, key)
|
|
except SourceBlockedError:
|
|
default_router.record_blocked(market, source_id)
|
|
res = None
|
|
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
|
|
per_market[market] = got
|
|
|
|
# Canonical ER is tenant-bound. Missing tenant context must never be
|
|
# papered over with a synthetic tenant or the removed Python scorer.
|
|
if tenant_id:
|
|
deduped = await resolve(
|
|
records,
|
|
tenant_id=tenant_id,
|
|
project_id=project_id,
|
|
correlation_id=correlation_id,
|
|
)
|
|
er_mode = "canonical_ts"
|
|
else:
|
|
deduped = [dict(r) for r in records]
|
|
er_mode = "fail_closed_unmerged_missing_tenant"
|
|
|
|
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
|
|
|
|
artifact_ref = f"0pt:pipeline_delivery:{order_id}"
|
|
for r in delivered:
|
|
for p in r.get("provenance", []):
|
|
p["artifact_ref"] = artifact_ref
|
|
|
|
quality_bridge_metrics: dict[str, Any]
|
|
if not tenant_id:
|
|
quality_bridge_metrics = {
|
|
"status": "not_configured",
|
|
"detail": "config.tenant_id missing — quality bridge not called",
|
|
}
|
|
else:
|
|
bridge_result = await run_quality_bridge(
|
|
tenant_id=tenant_id,
|
|
project_id=project_id,
|
|
dataset_id=f"0pt-order-{order_id}",
|
|
rows=delivered,
|
|
correlation_id=correlation_id,
|
|
)
|
|
quality_bridge_metrics = bridge_result.as_metrics()
|
|
if bridge_result.status == "ok":
|
|
quality_info = {
|
|
"score": bridge_result.score,
|
|
"run_id": bridge_result.run_id,
|
|
"run_status": bridge_result.run_status,
|
|
"checked_by": "0pt-quality-bridge",
|
|
}
|
|
for r in delivered:
|
|
for p in r.get("provenance", []):
|
|
p["quality"] = quality_info
|
|
elif bridge_result.status != "disabled":
|
|
logger.warning(
|
|
"0pt.quality_bridge did not complete for order %s: %s (%s)",
|
|
order_id, bridge_result.status, bridge_result.detail,
|
|
)
|
|
|
|
return {
|
|
"stage": "qa",
|
|
"pct": 100,
|
|
"state": "done",
|
|
"metrics": {
|
|
"rows": len(delivered),
|
|
"dropped_below_threshold": dropped,
|
|
"per_market": per_market,
|
|
"avg_confidence": avg_conf,
|
|
"pipeline_version": PIPELINE_VERSION,
|
|
"entity_resolution": er_mode,
|
|
"quality_bridge": quality_bridge_metrics,
|
|
},
|
|
"records": delivered,
|
|
}
|