188 lines
6.4 KiB
Python
188 lines
6.4 KiB
Python
"""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 only applies the returned canonical
|
|
`match` decisions. If the canonical engine is unavailable, records are
|
|
returned UNMERGED (fail closed against false identity merges).
|
|
"""
|
|
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 _apply_matches(records: list[dict[str, Any]], pairs: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
if not records:
|
|
return []
|
|
|
|
ids = [_record_id(r) for r in records]
|
|
# Exact duplicate payloads are indistinguishable; deterministic suffixes
|
|
# make request ids unique without changing merge semantics.
|
|
seen: dict[str, int] = {}
|
|
unique_ids: list[str] = []
|
|
for rid in ids:
|
|
n = seen.get(rid, 0)
|
|
seen[rid] = n + 1
|
|
unique_ids.append(rid if n == 0 else f"{rid}_{n}")
|
|
|
|
parent = {rid: rid for rid in unique_ids}
|
|
|
|
def find(x: str) -> str:
|
|
while parent[x] != x:
|
|
parent[x] = parent[parent[x]]
|
|
x = parent[x]
|
|
return x
|
|
|
|
def union(a: str, b: str) -> None:
|
|
ra, rb = find(a), find(b)
|
|
if ra != rb:
|
|
parent[max(ra, rb)] = min(ra, rb)
|
|
|
|
match_pairs = [p for p in pairs if p.get("status") == "match"]
|
|
for pair in match_pairs:
|
|
a, b = pair.get("record_a"), pair.get("record_b")
|
|
if a in parent and b in parent:
|
|
union(a, b)
|
|
|
|
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 members in groups.values():
|
|
members.sort(key=lambda item: (-float(item[1].get("confidence", 0.0)), item[0]))
|
|
anchor_id, anchor = members[0]
|
|
sources = sorted({str(r.get("source")) for _, r in members 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 members[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 members[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 members}
|
|
edges = [
|
|
p for p in match_pairs
|
|
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(members) > 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:
|
|
return [copy.deepcopy(r) for r in records]
|
|
|
|
ids = [_record_id(r) for r in records]
|
|
seen: dict[str, int] = {}
|
|
er_records = []
|
|
for rid, record in zip(ids, records):
|
|
n = seen.get(rid, 0)
|
|
seen[rid] = n + 1
|
|
unique = rid if n == 0 else f"{rid}_{n}"
|
|
name = str(record.get("name") or "")
|
|
er_records.append({
|
|
"record_id": unique,
|
|
"fields": {
|
|
"market": record.get("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"]
|