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

"""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,
    }