diff --git a/services/api/workers/resolve.py b/services/api/workers/resolve.py index ae50623..edb6161 100644 --- a/services/api/workers/resolve.py +++ b/services/api/workers/resolve.py @@ -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"]