236 lines
8.1 KiB
Python
236 lines
8.1 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 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"]
|