0pt/services/api/workers/resolve.py

"""
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.

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).
"""
from __future__ import annotations

from typing import Any

from rapidfuzz import fuzz


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 resolve(records: list[dict[str, Any]], threshold: int = 92) -> list[dict[str, Any]]:
    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", []):
                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:
                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