"""Thin adapter from the Python pipeline to 0pt's canonical TS ER engine. Python no longer scores identity. It sends opaque records to /api/0pt/internal/er/candidates, then applies only explicit hard-rule `match` decisions. Score/fuzzy decisions are review evidence only and can never merge records in this adapter, even if an older or compromised bridge returns `status=match, decided_by=score`. If the canonical engine is unavailable, records are returned UNMERGED (fail closed against false identity merges). A canonical hard non-match is also a cluster-wide constraint: transitive matches are never allowed to pull two records with conflicting stable identity evidence into the same entity. """ from __future__ import annotations import copy import hashlib import json import os from typing import Any import httpx _TIMEOUT = httpx.Timeout(6.0, connect=2.0) def _url() -> str | None: base = os.getenv("OPT_ER_BRIDGE_URL") or os.getenv("EDGE_URL") return f"{base.rstrip('/')}/api/0pt/internal/er/candidates" if base else None def _record_id(record: dict[str, Any]) -> str: stable = { "market": record.get("market"), "name": record.get("name"), "status": record.get("status"), "lei": record.get("lei"), "org_nr": record.get("org_nr"), "source": record.get("source"), "provenance": record.get("provenance", []), } raw = json.dumps(stable, sort_keys=True, separators=(",", ":"), default=str).encode() return "erp_" + hashlib.sha256(raw).hexdigest()[:32] def _unique_record_ids(records: list[dict[str, Any]]) -> list[str]: seen: dict[str, int] = {} unique_ids: list[str] = [] for record in records: rid = _record_id(record) n = seen.get(rid, 0) seen[rid] = n + 1 unique_ids.append(rid if n == 0 else f"{rid}_{n}") return unique_ids def _apply_matches(records: list[dict[str, Any]], pairs: list[dict[str, Any]]) -> list[dict[str, Any]]: if not records: return [] unique_ids = _unique_record_ids(records) known_ids = set(unique_ids) parent = {rid: rid for rid in unique_ids} members = {rid: {rid} for rid in unique_ids} hard_non_matches: set[frozenset[str]] = set() for pair in pairs: a, b = pair.get("record_a"), pair.get("record_b") if a not in known_ids or b not in known_ids: continue if pair.get("status") == "non_match" and pair.get("decided_by") not in (None, "score"): hard_non_matches.add(frozenset((a, b))) def find(x: str) -> str: while parent[x] != x: parent[x] = parent[parent[x]] x = parent[x] return x def components_conflict(left: set[str], right: set[str]) -> bool: return any(frozenset((a, b)) in hard_non_matches for a in left for b in right) def union(a: str, b: str) -> bool: ra, rb = find(a), find(b) if ra == rb: return True left, right = members[ra], members[rb] if components_conflict(left, right): return False keep, drop = (ra, rb) if ra < rb else (rb, ra) parent[drop] = keep members[keep] = members[ra] | members[rb] members.pop(drop, None) return True # Defence in depth: only stable-identifier hard rules may create canonical # identity links. Score/fuzzy "match" responses are ignored at the consumer # boundary even if an upstream/legacy route emits one. match_pairs = [ p for p in pairs if p.get("status") == "match" and p.get("decided_by") not in (None, "score") and p.get("record_a") in known_ids and p.get("record_b") in known_ids ] match_pairs.sort( key=lambda p: ( -float(p.get("score") or 0.0), str(p.get("candidate_id") or ""), ) ) accepted_edges: list[dict[str, Any]] = [] for pair in match_pairs: a, b = str(pair["record_a"]), str(pair["record_b"]) if union(a, b): accepted_edges.append(pair) groups: dict[str, list[tuple[str, dict[str, Any]]]] = {} for rid, record in zip(unique_ids, records): groups.setdefault(find(rid), []).append((rid, copy.deepcopy(record))) out: list[dict[str, Any]] = [] for group in groups.values(): group.sort(key=lambda item: (-float(item[1].get("confidence", 0.0)), item[0])) _, anchor = group[0] sources = sorted({str(r.get("source")) for _, r in group if r.get("source")}) anchor["agreed_sources"] = len(sources) or 1 anchor["confidence"] = min( 0.99, float(anchor.get("confidence", 0.0)) + 0.1 * max(0, len(sources) - 1), ) anchor["merged_from"] = [r.get("source") for _, r in group[1:] if r.get("source")] provenance = list(anchor.get("provenance", [])) keys = {(p.get("field_name"), p.get("source_id")) for p in provenance} for _, record in group[1:]: for p in record.get("provenance", []): key = (p.get("field_name"), p.get("source_id")) if key not in keys: provenance.append(copy.deepcopy(p)) keys.add(key) member_ids = {rid for rid, _ in group} edges = [ p for p in accepted_edges if p.get("record_a") in member_ids and p.get("record_b") in member_ids ] er_info = { "method": "canonical_ts_entity_resolution", "merged_sources": sources, "matched_edges": [ { "candidate_id": p.get("candidate_id"), "score": p.get("score"), "decided_by": p.get("decided_by"), } for p in edges ], } if len(group) > 1: for p in provenance: p["entity_resolution"] = er_info anchor["provenance"] = provenance out.append(anchor) return sorted(out, key=lambda r: (_record_id(r), str(r.get("source") or ""))) async def resolve( records: list[dict[str, Any]], *, tenant_id: str, project_id: str, correlation_id: str | None = None, ) -> list[dict[str, Any]]: """Return canonically resolved records; never falls back to local scoring.""" if len(records) < 2: return [copy.deepcopy(r) for r in records] url = _url() token = os.getenv("INTERNAL_TOKEN", "") if not url or not token or not tenant_id or not project_id: return [copy.deepcopy(r) for r in records] unique_ids = _unique_record_ids(records) er_records = [] for rid, record in zip(unique_ids, records): name = str(record.get("name") or "") market = record.get("market") er_records.append({ "record_id": rid, "fields": { "market": market, "registration_country": market, "block_initial": name[:1].upper(), "name": name, "status": record.get("status"), "lei": record.get("lei"), "org_nr": record.get("org_nr"), }, }) payload: dict[str, Any] = { "tenant_id": tenant_id, "project_id": project_id, "records": er_records, "blocking_fields": ["market", "block_initial"], "compare_fields": [{"field": "name", "weight": 1}], "auto_match_threshold": 0.92, "review_threshold": 0.72, } if correlation_id: payload["correlation_id"] = correlation_id try: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: response = await client.post( url, json=payload, headers={"Content-Type": "application/json", "X-Internal-Token": token}, ) if response.status_code not in (200, 201): return [copy.deepcopy(r) for r in records] body = response.json() pairs = body.get("pairs") if not isinstance(pairs, list): return [copy.deepcopy(r) for r in records] return _apply_matches(records, pairs) except (httpx.HTTPError, ValueError, TypeError, KeyError): return [copy.deepcopy(r) for r in records] __all__ = ["resolve", "_apply_matches"]