refactor(er): remove duplicate Python scoring engine
This commit is contained in:
+174
-104
@@ -1,117 +1,187 @@
|
||||
"""
|
||||
Entity resolution (docs/06): block, score, and link records across sources so the
|
||||
same company from GLEIF + Brreg + CVR collapses to one record with corroborated
|
||||
confidence. rapidfuzz only; deterministic; no external calls.
|
||||
"""Thin adapter from the Python pipeline to 0pt's canonical TS ER engine.
|
||||
|
||||
=======================================================================
|
||||
P065 — THIS IS THE SECOND ER ENGINE. IT IS NOT THE CANONICAL ONE.
|
||||
=======================================================================
|
||||
0pt has TWO entity-resolution implementations, and this is the simplified
|
||||
one:
|
||||
|
||||
* CANONICAL: src/er/entity-resolution-service.ts — the declared generic ER
|
||||
capability. Deterministic blocking + an explainable per-pair score,
|
||||
product-supplied HARD RULES that override the score in both directions,
|
||||
a configurable review band that hands ambiguous pairs to humans instead
|
||||
of auto-merging, and delegation of heavy probabilistic linkage (Splink)
|
||||
to 0pt-motor via the N0D job "0pt.er.batch_resolution".
|
||||
|
||||
* THIS FILE: blocking on (market, first initial) + a single
|
||||
rapidfuzz token_sort_ratio threshold, first-match-wins, auto-merge with
|
||||
no review band, no hard-rule hook, no explanation beyond the score.
|
||||
|
||||
P065's definition of done is that no second simplified production engine
|
||||
remains. It has NOT been met: this file is still what
|
||||
`workers/pipeline.py` calls (line ~110, `deduped = resolve(records)`).
|
||||
|
||||
WHY IT WAS NOT REPLACED IN THE P060-P068 PASS, stated plainly rather than
|
||||
quietly deferred: replacing it means turning this module into an ADAPTER
|
||||
that calls the canonical capability, and that change cannot be verified
|
||||
from this repo today —
|
||||
* `services/api` is not deployed by this repo's Dockerfile (which builds
|
||||
only the Next.js app), and api.0pt.app answers 404, so there is no
|
||||
running engine to test against;
|
||||
* no CI workflow runs services/api's pytest suite at all (verified:
|
||||
`grep -rn "pytest" .gitea/workflows .github/workflows` finds nothing),
|
||||
so a rewrite here would land with zero automated coverage.
|
||||
Rewriting it blind and calling P065 done would be exactly the
|
||||
"claimed complete, broken in operation" failure this programme exists to
|
||||
stop. The removal plan is recorded in STATUS.md.
|
||||
|
||||
Until then: do NOT add capability to this file. Any new ER behaviour
|
||||
belongs in the canonical TypeScript service.
|
||||
|
||||
Deterministic by construction: for a fixed `records` input, blocking is a pure
|
||||
function of (market, name-first-letter) and matching is first-match-wins
|
||||
within a block, so results do not depend on which source answered first — see
|
||||
tests/p0/test_resolve_determinism.py, which asserts this holds across
|
||||
shuffled input orderings (BEFORE_STATE.md §4 flagged this as untested; it is
|
||||
order-SENSITIVE only in the sense that ties within a block resolve to
|
||||
whichever candidate appears earliest in `out`, which is itself deterministic
|
||||
given a fixed `records` order — the test pins exactly that contract).
|
||||
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
|
||||
|
||||
from rapidfuzz import fuzz
|
||||
import httpx
|
||||
|
||||
_TIMEOUT = httpx.Timeout(6.0, connect=2.0)
|
||||
|
||||
|
||||
def block_key(rec: dict[str, Any]) -> tuple[str, str]:
|
||||
"""Cheap blocking key — same market + first initial — to avoid O(n^2)."""
|
||||
return (rec.get("market") or "", (rec.get("name") or "")[:1].upper())
|
||||
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 resolve(records: list[dict[str, Any]], threshold: int = 92) -> list[dict[str, Any]]:
|
||||
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 r in records:
|
||||
match = None
|
||||
match_score: float | None = None
|
||||
for o in out:
|
||||
if block_key(o) != block_key(r):
|
||||
continue
|
||||
score = fuzz.token_sort_ratio(o.get("name", ""), r.get("name", ""))
|
||||
if score >= threshold:
|
||||
match = o
|
||||
match_score = score
|
||||
break
|
||||
if match:
|
||||
match.setdefault("merged_from", []).append(r.get("source"))
|
||||
# Merge the losing record's per-field provenance regardless of source
|
||||
# novelty below, so the surviving record's provenance always reflects
|
||||
# every contributing record's fields.
|
||||
existing_provenance = match.setdefault("provenance", [])
|
||||
seen_provenance = {(p.get("field_name"), p.get("source_id")) for p in existing_provenance}
|
||||
for p in r.get("provenance", []):
|
||||
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 seen_provenance:
|
||||
existing_provenance.append(p)
|
||||
seen_provenance.add(key)
|
||||
# Corroboration (confidence/agreed_sources bump) requires an
|
||||
# *independent* source — two hits from the same connector must not
|
||||
# count as two agreeing sources.
|
||||
seen_sources = match.setdefault("_sources_seen", {match.get("source")} - {None})
|
||||
src = r.get("source")
|
||||
if src is not None and src not in seen_sources:
|
||||
match["confidence"] = min(0.99, match.get("confidence", 0.0) + 0.1)
|
||||
match["agreed_sources"] = match.get("agreed_sources", 1) + 1
|
||||
seen_sources.add(src)
|
||||
# Record the entity-resolution decision on every provenance row the
|
||||
# surviving record now carries (BEFORE_STATE.md §3: no per-field
|
||||
# entity_resolution tracking existed anywhere in the repo). Rows on
|
||||
# a record that was never merged into anything keep
|
||||
# entity_resolution == None — honest, not backfilled with a guess.
|
||||
er_info = {
|
||||
"method": "blocking+fuzzy_token_sort_ratio",
|
||||
"block_key": list(block_key(match)),
|
||||
"match_score": match_score,
|
||||
"threshold": threshold,
|
||||
"merged_sources": sorted(s for s in match["_sources_seen"] if s is not None),
|
||||
}
|
||||
for p in existing_provenance:
|
||||
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
|
||||
else:
|
||||
out.append({**r, "agreed_sources": 1, "_sources_seen": {r.get("source")} - {None}})
|
||||
for o in out:
|
||||
o.pop("_sources_seen", None)
|
||||
return out
|
||||
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"]
|
||||
|
||||
Reference in New Issue
Block a user