From 78b161f56df2160c02b3c1bc6631f7563221f106 Mon Sep 17 00:00:00 2001 From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com> Date: Tue, 15 Sep 2026 13:48:53 +0530 Subject: [PATCH] fix(er): make the clerical review loop round-trip and keep cluster identity load_review_decisions (#239): CSV queues were read with type inference, so ids such as "007" came back as 7 and never matched their pair, and a blank decision cell became NaN -> 'nan' and raised ValueError. The CSV is now read as text (dtype=str, keep_default_na=False); _coerce_id maps None, NaN/pd.NA and "" to None, and blank decision cells (in any format) are treated as undecided and skipped. load_review_decisions (#240): the CSV export's formula sanitizer also guards the id columns, so "+4410" is written as "'+4410" and the loaded key never matched. On CSV load only, a leading ' is stripped from left_id/right_id/item_id when the remainder starts with a formula prefix (the read-only _FORMULA_PREFIXES). The export path and the sanitizer are unchanged; JSONL/parquet ids are read verbatim because those formats are never sanitized. apply_review_decisions (#267): decisions carrying only an item_id were silently dropped. A new keyword-only queue= parameter resolves item ids through the ReviewQueueReport the reviewer worked from (item ids are queue positions that depend on the queue config, so they are not re-derived from the report). A decision that cannot be resolved to a pair now raises ValueError naming it. feedback_summary gains n_unmatched, the number of decided pairs absent from the report, and recalibration uses the resolved decisions. _recluster_from_pairs (#268): clusters were renumbered from er_000000 over matched records only and the canonical record reset to the lowest id, even with no decisions. The report's clusters are now passed in: an unchanged component keeps its cluster exactly; otherwise an original cluster's id and canonical record follow its canonical record (a merge keeps the lowest original id); a component with no original canonical gets a fresh id numbered past n_records and every existing er_N id. Closes #239 Closes #240 Closes #267 Closes #268 --- src/freshdata/enterprise/entity_resolution.py | 163 ++++++- tests/test_er_review_loop.py | 442 ++++++++++++++++++ 2 files changed, 586 insertions(+), 19 deletions(-) create mode 100644 tests/test_er_review_loop.py diff --git a/src/freshdata/enterprise/entity_resolution.py b/src/freshdata/enterprise/entity_resolution.py index c768e5a4..a0202778 100644 --- a/src/freshdata/enterprise/entity_resolution.py +++ b/src/freshdata/enterprise/entity_resolution.py @@ -42,7 +42,7 @@ import pandas as pd -from .._util import sanitize_csv_formulas +from .._util import _FORMULA_PREFIXES, sanitize_csv_formulas from ..adapters.polars import from_pandas, to_pandas from .config import ( # noqa: F401 (configs re-exported for discoverability) BlockingRule, @@ -1631,15 +1631,41 @@ def export_review_queue( def _coerce_id(value: Any) -> Any: + """Normalize a loaded cell: ``None``, NaN/NA and ``""`` all mean "not given".""" if value is None: return None - if isinstance(value, float) and pd.isna(value): + if isinstance(value, str): + return value if value != "" else None + if pd.api.types.is_scalar(value) and pd.isna(value): return None return value +def _strip_formula_guard(value: Any) -> Any: + """Undo the ``'`` that CSV formula sanitization prepends to a cell. + + Only a ``'`` followed by a formula prefix is removed, so a loaded key + matches the id it was exported from. This is the inverse of the export-side + guard, not a weakening of it: nothing loaded here is written back to a + spreadsheet. + """ + if ( + isinstance(value, str) + and value.startswith("'") + and value[1:].lstrip(" \t").startswith(_FORMULA_PREFIXES) + ): + return value[1:] + return value + + def load_review_decisions(path: str | Path, *, format: str | None = None) -> list[ReviewDecision]: - """Load clerical decisions written back by a reviewer (csv/jsonl/parquet).""" + """Load clerical decisions written back by a reviewer (csv/jsonl/parquet). + + Rows whose ``decision`` cell is blank are undecided and skipped. CSV cells + are read as text, so ids such as ``"007"`` keep their exact spelling, and + the ``'`` guard that :func:`export_review_queue` adds to formula-like ids + (``'+4410``) is removed again from ``left_id``/``right_id``/``item_id``. + """ src = Path(path) fmt = _resolve_format(src, format) rows: list[dict[str, Any]] @@ -1651,24 +1677,31 @@ def load_review_decisions(path: str | Path, *, format: str | None = None) -> lis if text: rows.append(json.loads(text)) elif fmt == "csv": - rows = pd.read_csv(src).to_dict("records") + rows = pd.read_csv(src, dtype=str, keep_default_na=False).to_dict("records") else: # parquet rows = pd.read_parquet(src).to_dict("records") + def key_cell(row: dict[str, Any], name: str) -> Any: + value = _coerce_id(row.get(name)) + # Only the CSV export applies the formula guard. + return _strip_formula_guard(value) if fmt == "csv" else value + decisions: list[ReviewDecision] = [] for row in rows: - decision = str(row.get("decision", "")).strip() + raw_decision = _coerce_id(row.get("decision")) + decision = "" if raw_decision is None else str(raw_decision).strip() if not decision: continue + note = _coerce_id(row.get("note")) decisions.append( ReviewDecision( decision=decision, # type: ignore[arg-type] - left_id=_coerce_id(row.get("left_id")), - right_id=_coerce_id(row.get("right_id")), - item_id=_coerce_id(row.get("item_id")), + left_id=key_cell(row, "left_id"), + right_id=key_cell(row, "right_id"), + item_id=key_cell(row, "item_id"), reviewer=_coerce_id(row.get("reviewer")), decided_at=_coerce_id(row.get("decided_at")), - note=str(row.get("note") or ""), + note="" if note is None else str(note), ) ) return decisions @@ -1679,8 +1712,31 @@ def load_review_decisions(path: str | Path, *, format: str | None = None) -> lis # ===================================================================== -def _recluster_from_pairs(pairs: list[MatchPair], n_records: int) -> list[EntityCluster]: - """Rebuild multi-record clusters from match pairs alone (frame-independent).""" +def _cluster_id_order(cluster_id: str) -> tuple[int, str]: + # "er_000009" < "er_000010" < "er_1000000": fixed-width ids widen past 6 digits. + return (len(cluster_id), cluster_id) + + +def _recluster_from_pairs( + pairs: list[MatchPair], + n_records: int, + original: Sequence[EntityCluster] = (), +) -> list[EntityCluster]: + """Rebuild multi-record clusters from match pairs alone (frame-independent). + + Cluster identity carries over from *original* (the report's clusters), so + folding decisions back in never renumbers untouched entities: + + * a component whose membership equals an original cluster is returned as + that cluster (same id, record order, canonical record and confidence); + * otherwise an original cluster's ``cluster_id`` and canonical record pass + to the component that now holds that canonical record; when a merge + holds several, the lowest original ``cluster_id`` wins; + * a component holding no original canonical record gets a fresh id + numbered past ``n_records`` and every existing ``er_N`` id (so it cannot + collide with a ``cluster_id`` in the resolved frame), and, with no frame + to score completeness, its smallest record id as canonical. + """ from collections import defaultdict parent: dict[str, str] = {} @@ -1714,30 +1770,87 @@ def union(a: str, b: str) -> None: for p in matched: conf[find(str(p.left_id))].append(p.match_probability) - ordered = sorted(members.values(), key=min) + same_members = {frozenset(str(r) for r in c.record_ids): c for c in original} + heir_of: dict[str, EntityCluster] = {} + next_idx = n_records + for c in sorted(original, key=lambda c: _cluster_id_order(c.cluster_id)): + heir_of.setdefault(str(c.canonical_record_id), c) + numbered = re.fullmatch(r"er_(\d+)", c.cluster_id) + if numbered: + next_idx = max(next_idx, int(numbered.group(1)) + 1) + clusters: list[EntityCluster] = [] - for idx, keys in enumerate(ordered): + for keys in sorted(members.values(), key=min): if len(keys) < 2: continue + unchanged = same_members.get(frozenset(keys)) + if unchanged is not None: + clusters.append(replace(unchanged)) + continue + heirs = [heir_of[k] for k in keys if k in heir_of] + if heirs: + heir = min(heirs, key=lambda c: _cluster_id_order(c.cluster_id)) + cluster_id, canonical = heir.cluster_id, heir.canonical_record_id + else: + cluster_id, canonical = f"er_{next_idx:06d}", id_of[min(keys)] + next_idx += 1 confs = conf.get(find(keys[0]), []) clusters.append( EntityCluster( - cluster_id=f"er_{idx:06d}", + cluster_id=cluster_id, record_ids=tuple(id_of[k] for k in sorted(keys)), size=len(keys), - canonical_record_id=id_of[min(keys)], + canonical_record_id=canonical, confidence=sum(confs) / len(confs) if confs else 1.0, ) ) return clusters +def _resolve_decision_pairs( + decisions: Sequence[ReviewDecision], + queue: ReviewQueueReport | None, +) -> list[ReviewDecision]: + """Return *decisions* with every one addressable by ``(left_id, right_id)``. + + A decision without a complete pair is looked up by ``item_id`` in *queue*. + Item ids are positions in the queue the reviewer worked from, so they are + never re-derived from the report; an unresolvable decision raises. + """ + pair_of = {} if queue is None else {it.item_id: it for it in queue.items} + resolved: list[ReviewDecision] = [] + unresolved: list[str] = [] + for d in decisions: + if d.pair_key is not None: + resolved.append(d) + elif d.item_id is not None and d.item_id in pair_of: + item = pair_of[d.item_id] + resolved.append(replace(d, left_id=item.left_id, right_id=item.right_id)) + elif d.item_id is not None: + unresolved.append(f"item_id={d.item_id!r}") + else: + unresolved.append(f"left_id={d.left_id!r}, right_id={d.right_id!r}") + if unresolved: + hint = ( + "pass queue= (the ReviewQueueReport the reviewer worked from) or give " + "left_id and right_id" + if queue is None + else "these item ids are not in the given queue" + ) + raise ValueError( + f"cannot resolve {len(unresolved)} review decision(s) to a pair " + f"({'; '.join(unresolved)}): {hint}" + ) + return resolved + + def apply_review_decisions( report: EntityResolutionReport, decisions: Sequence[ReviewDecision], *, config: EntityResolutionConfig | None = None, recalibrate: bool = False, + queue: ReviewQueueReport | None = None, ) -> EntityResolutionReport: """Fold clerical decisions back into a report. @@ -1746,21 +1859,32 @@ def apply_review_decisions( cluster. Returns a **new** report (the input is not mutated). Config is only recalibrated when ``recalibrate=True`` *and* a ``config`` is supplied — the safe default leaves your config untouched. + + A decision that carries only an ``item_id`` is resolved through *queue*, + the :class:`ReviewQueueReport` the reviewer worked from; ``ValueError`` is + raised when a decision cannot be resolved to a pair. + ``feedback_summary["n_unmatched"]`` counts decided pairs that match no pair + in *report*. Clusters keep their ``cluster_id`` and canonical record unless + the decisions change their membership (see :func:`_recluster_from_pairs`). """ + resolved = _resolve_decision_pairs(decisions, queue) by_key: dict[tuple[str, str], ReviewDecision] = {} - for d in decisions: + for d in resolved: key = d.pair_key if key is not None: by_key[key] = d counts = {"accept": 0, "reject": 0, "manual_merge": 0} promoted = demoted = 0 + seen_keys: set[tuple[str, str]] = set() new_pairs: list[MatchPair] = [] for p in report.pairs: - dec = by_key.get(_pair_key(p.left_id, p.right_id)) + pair_key = _pair_key(p.left_id, p.right_id) + dec = by_key.get(pair_key) if dec is None: new_pairs.append(p) continue + seen_keys.add(pair_key) counts[dec.decision] += 1 before = p.decision if dec.decision in ("accept", "manual_merge"): @@ -1773,16 +1897,17 @@ def apply_review_decisions( demoted += 1 new_pairs.append(replace(p, decision=after)) - clusters = _recluster_from_pairs(new_pairs, report.n_records) + clusters = _recluster_from_pairs(new_pairs, report.n_records, report.clusters) feedback = { "decisions": counts, "n_applied": sum(counts.values()), + "n_unmatched": len(by_key.keys() - seen_keys), "n_promoted": promoted, "n_demoted": demoted, "updated_at": _utcnow_iso(), } if recalibrate and config is not None: - new_config = recalibrate_weights(config, report, decisions) + new_config = recalibrate_weights(config, report, resolved) feedback["recalibrated_weights"] = { c.column: round(c.weight, 4) for c in new_config.comparisons } diff --git a/tests/test_er_review_loop.py b/tests/test_er_review_loop.py new file mode 100644 index 00000000..6f9c3706 --- /dev/null +++ b/tests/test_er_review_loop.py @@ -0,0 +1,442 @@ +"""Regression tests for the clerical review loop: export -> reviewer edit -> +load_review_decisions -> apply_review_decisions (#239, #240, #267, #268).""" + +from __future__ import annotations + +import json + +import numpy as np +import pandas as pd +import pytest + +from freshdata.enterprise import ( + BlockingRule, + ComparisonLevel, + EntityCluster, + EntityResolutionConfig, + MatchPair, + ReviewDecision, + apply_review_decisions, + build_review_queue, + export_review_queue, + load_review_decisions, + resolve_entities, +) +from freshdata.enterprise.entity_resolution import ( + EntityResolutionReport, + _coerce_id, + _strip_formula_guard, +) + + +def _review_config() -> EntityResolutionConfig: + # match_threshold=0.99 puts near-identical names in the clerical-review band. + return EntityResolutionConfig( + enabled=True, + backend="pandas", + blocking_rules=(BlockingRule("l.e = r.e"),), + comparisons=(ComparisonLevel("n", "jaro_winkler"),), + match_threshold=0.99, + clerical_review_threshold=0.5, + ) + + +def _review_report(ids: list[str]) -> EntityResolutionReport: + df = pd.DataFrame( + { + "id": ids, + "e": ["x", "x", "y", "y"][: len(ids)], + "n": ["ann", "anne", "bob", "bobb"][: len(ids)], + } + ) + return resolve_entities(df, config=_review_config())[1] + + +def _reviewer_fills_csv(path, decisions) -> None: + """Simulate a reviewer editing the exported queue as text.""" + q = pd.read_csv(path, dtype=str, keep_default_na=False) + q["decision"] = decisions + q.to_csv(path, index=False) + + +# --------------------------------------------------------------------------- # +# #239: CSV round-trip keeps id spelling; blank decision cells are undecided +# --------------------------------------------------------------------------- # + + +def test_csv_roundtrip_keeps_leading_zero_ids(tmp_path): + report = _review_report(["007", "008"]) + path = export_review_queue(report, tmp_path / "queue.csv") + _reviewer_fills_csv(path, "accept") + + decisions = load_review_decisions(path) + assert [(d.left_id, d.right_id) for d in decisions] == [("007", "008")] + + updated = apply_review_decisions(report, decisions) + assert updated.feedback_summary["n_applied"] == 1 + assert updated.feedback_summary["n_unmatched"] == 0 + assert updated.n_matches == 1 + + +def test_csv_blank_decision_cells_are_skipped(tmp_path): + report = _review_report(["a", "b", "c", "d"]) + path = export_review_queue(report, tmp_path / "queue.csv") + q = pd.read_csv(path) + assert len(q) == 2 + q["decision"] = ["accept", None] # the reviewer decided one row only + q.to_csv(path, index=False) + + decisions = load_review_decisions(path) + assert len(decisions) == 1 + assert decisions[0].decision == "accept" + # Blank optional cells come back as None, not "" or NaN. + assert decisions[0].reviewer is None + assert decisions[0].decided_at is None + assert decisions[0].note == "" + assert apply_review_decisions(report, decisions).feedback_summary["n_applied"] == 1 + + +def test_csv_whitespace_only_decision_is_skipped(tmp_path): + path = tmp_path / "d.csv" + pd.DataFrame( + [ + {"left_id": "1", "right_id": "2", "decision": " "}, + {"left_id": "3", "right_id": "4", "decision": "reject"}, + ] + ).to_csv(path, index=False) + assert [d.decision for d in load_review_decisions(path)] == ["reject"] + + +def test_jsonl_null_decision_is_skipped(tmp_path): + path = tmp_path / "d.jsonl" + rows = [ + {"left_id": 1, "right_id": 2, "decision": None, "note": None}, + {"left_id": 3, "right_id": 4, "decision": "accept", "note": None}, + ] + path.write_text("\n".join(json.dumps(r) for r in rows)) + decisions = load_review_decisions(path) + assert [(d.left_id, d.decision, d.note) for d in decisions] == [(3, "accept", "")] + + +def test_parquet_null_decision_is_skipped(tmp_path): + pytest.importorskip("pyarrow") + path = tmp_path / "d.parquet" + pd.DataFrame( + { + "left_id": [1, 3], + "right_id": [2, 4], + "decision": [None, "accept"], + "reviewer": [None, None], + } + ).to_parquet(path, index=False) + decisions = load_review_decisions(path) + assert len(decisions) == 1 + assert decisions[0].decision == "accept" + assert decisions[0].reviewer is None + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (None, None), + (float("nan"), None), + (np.nan, None), + (pd.NA, None), + ("", None), + ("007", "007"), + (" ", " "), + (0, 0), + (7, 7), + ], +) +def test_coerce_id(value, expected): + assert _coerce_id(value) is expected or _coerce_id(value) == expected + + +# --------------------------------------------------------------------------- # +# #240: the CSV formula guard on id cells is undone on load +# --------------------------------------------------------------------------- # + + +@pytest.mark.parametrize("prefix", ["+44", "-", "=", "@", " =", "\t="]) +def test_csv_roundtrip_formula_like_ids(tmp_path, prefix): + ids = [f"{prefix}10", f"{prefix}11"] + report = _review_report(ids) + path = export_review_queue(report, tmp_path / "queue.csv") + # The export-side sanitizer is unchanged: the ids are still guarded on disk. + assert f"'{ids[0]}" in path.read_text(encoding="utf-8") + _reviewer_fills_csv(path, "accept") + + decisions = load_review_decisions(path) + assert {decisions[0].left_id, decisions[0].right_id} == set(ids) + updated = apply_review_decisions(report, decisions) + assert updated.feedback_summary["n_applied"] == 1 + assert updated.n_matches == 1 + + +def test_csv_unsanitized_export_roundtrip(tmp_path): + report = _review_report(["+4410", "+4411"]) + path = export_review_queue(report, tmp_path / "queue.csv", sanitize_formulas=False) + _reviewer_fills_csv(path, "accept") + decisions = load_review_decisions(path) + assert decisions[0].left_id == "+4410" + assert apply_review_decisions(report, decisions).feedback_summary["n_applied"] == 1 + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("'+4410", "+4410"), + ("'=SUM(A1)", "=SUM(A1)"), + ("' =1", " =1"), + ("'abc", "'abc"), # not a guard: no formula prefix follows + ("'", "'"), + ("+4410", "+4410"), + (7, 7), + (None, None), + ], +) +def test_strip_formula_guard(value, expected): + assert _strip_formula_guard(value) == expected + + +def test_formula_guard_only_stripped_for_csv(tmp_path): + # JSONL is never sanitized on export, so a leading ' is part of the id. + path = tmp_path / "d.jsonl" + path.write_text(json.dumps({"left_id": "'+1", "right_id": "'+2", "decision": "accept"})) + decisions = load_review_decisions(path) + assert (decisions[0].left_id, decisions[0].right_id) == ("'+1", "'+2") + + +def test_csv_guard_not_stripped_from_free_text(tmp_path): + path = tmp_path / "d.csv" + pd.DataFrame( + [{"left_id": "1", "right_id": "2", "decision": "accept", "note": "'=cmd"}] + ).to_csv(path, index=False) + assert load_review_decisions(path)[0].note == "'=cmd" + + +# --------------------------------------------------------------------------- # +# #267: item_id-only decisions are resolved through the queue, or rejected +# --------------------------------------------------------------------------- # + + +def _two_record_report() -> EntityResolutionReport: + df = pd.DataFrame({"id": ["a", "b"], "e": ["x", "x"], "n": ["ann", "anne"]}) + return resolve_entities(df, config=_review_config())[1] + + +def test_item_id_only_decision_resolved_via_queue(): + report = _two_record_report() + queue = build_review_queue(report) + item = queue.items[0] + out = apply_review_decisions( + report, [ReviewDecision("accept", item_id=item.item_id)], queue=queue + ) + assert out.feedback_summary["n_applied"] == 1 + assert out.feedback_summary["n_promoted"] == 1 + assert out.n_matches == 1 + assert [set(c.record_ids) for c in out.clusters] == [{"a", "b"}] + + +def test_item_id_only_decision_without_queue_raises(): + report = _two_record_report() + item = build_review_queue(report).items[0] + with pytest.raises(ValueError, match=r"queue=") as exc: + apply_review_decisions(report, [ReviewDecision("accept", item_id=item.item_id)]) + assert item.item_id in str(exc.value) + + +def test_unknown_item_id_raises_even_with_queue(): + report = _two_record_report() + queue = build_review_queue(report) + with pytest.raises(ValueError, match="rev_999999"): + apply_review_decisions( + report, [ReviewDecision("accept", item_id="rev_999999")], queue=queue + ) + + +def test_incomplete_pair_without_item_id_raises(): + report = _two_record_report() + with pytest.raises(ValueError, match="left_id='a'"): + apply_review_decisions(report, [ReviewDecision("accept", left_id="a")]) + + +def test_pair_decision_wins_over_queue_lookup(): + report = _two_record_report() + queue = build_review_queue(report) + out = apply_review_decisions( + report, + [ReviewDecision("reject", left_id="b", right_id="a", item_id="rev_999999")], + queue=queue, + ) + assert out.feedback_summary["decisions"]["reject"] == 1 + + +def test_n_unmatched_counts_decisions_for_unknown_pairs(): + report = _two_record_report() + out = apply_review_decisions( + report, + [ + ReviewDecision("accept", left_id="a", right_id="b"), + ReviewDecision("accept", left_id="a", right_id="zzz"), + ], + ) + assert out.feedback_summary["n_applied"] == 1 + assert out.feedback_summary["n_unmatched"] == 1 + + +def test_recalibrate_uses_item_id_resolved_decisions(): + config = _review_config() + report = _two_record_report() + queue = build_review_queue(report) + out = apply_review_decisions( + report, + [ReviewDecision("accept", item_id=queue.items[0].item_id)], + config=config, + recalibrate=True, + queue=queue, + ) + assert out.feedback_summary["recalibrated_weights"]["n"] != config.comparisons[0].weight + + +# --------------------------------------------------------------------------- # +# #268: applying decisions keeps cluster identity +# --------------------------------------------------------------------------- # + + +def test_apply_no_decisions_is_identity_on_clusters(): + df = pd.DataFrame({"id": [1, 2, 3, 4], "e": ["a", "b", "b", "c"], "x": [1, None, 5, 1]}) + cfg = EntityResolutionConfig( + enabled=True, + backend="pandas", + blocking_rules=(BlockingRule("l.e = r.e"),), + comparisons=(ComparisonLevel("e"),), + ) + frame, report = resolve_entities(df, config=cfg) + after = apply_review_decisions(report, []) + + assert [c.to_dict() for c in after.clusters] == [c.to_dict() for c in report.clusters] + assert [(c.cluster_id, c.canonical_record_id) for c in after.clusters] == [("er_000001", 3)] + assert set(frame["cluster_id"]) >= {c.cluster_id for c in after.clusters} + assert after.n_clusters == report.n_clusters + # The returned clusters are copies; the input report is not shared. + assert all(a is not b for a, b in zip(after.clusters, report.clusters)) + + +def _pair(left, right, decision, score=0.9) -> MatchPair: + return MatchPair(left, right, score, score, {"f": score}, decision) + + +def _manual_report(pairs, clusters, n_records=8) -> EntityResolutionReport: + return EntityResolutionReport( + n_records=n_records, + n_candidate_pairs=len(pairs), + n_matches=sum(p.decision == "match" for p in pairs), + n_possible_matches=sum(p.decision == "possible_match" for p in pairs), + n_clusters=len(clusters), + backend="pandas", + pairs=pairs, + clusters=clusters, + ) + + +def _merge_report() -> EntityResolutionReport: + # Clusters {1,2} (canonical 2) and {3,4} (canonical 4); (2,3) links them; + # (5,6) and (7,8) are singletons in the review band. + pairs = [ + _pair(1, 2, "match", 0.99), + _pair(3, 4, "match", 0.97), + _pair(2, 3, "possible_match", 0.7), + _pair(5, 6, "possible_match", 0.7), + _pair(7, 8, "possible_match", 0.7), + ] + clusters = [ + EntityCluster("er_000000", (1, 2), 2, 2, 0.99), + EntityCluster("er_000002", (3, 4), 2, 4, 0.97), + ] + return _manual_report(pairs, clusters) + + +def test_redundant_accept_keeps_cluster_unchanged(): + report = _merge_report() + out = apply_review_decisions(report, [ReviewDecision("accept", left_id=2, right_id=1)]) + assert [c.to_dict() for c in out.clusters] == [c.to_dict() for c in report.clusters] + + +def test_merge_keeps_lowest_original_id_and_its_canonical(): + report = _merge_report() + out = apply_review_decisions(report, [ReviewDecision("accept", left_id=2, right_id=3)]) + assert len(out.clusters) == 1 + merged = out.clusters[0] + assert merged.cluster_id == "er_000000" + assert merged.canonical_record_id == 2 + assert set(merged.record_ids) == {1, 2, 3, 4} + assert merged.size == 4 + + +def test_new_cluster_ids_do_not_collide_across_rounds(): + report = _merge_report() + first = apply_review_decisions(report, [ReviewDecision("accept", left_id=5, right_id=6)]) + ids_first = [c.cluster_id for c in first.clusters] + assert ids_first == ["er_000000", "er_000002", "er_000008"] # n_records=8 + + second = apply_review_decisions(first, [ReviewDecision("accept", left_id=7, right_id=8)]) + ids_second = [c.cluster_id for c in second.clusters] + assert ids_second == ["er_000000", "er_000002", "er_000008", "er_000009"] + assert len(set(ids_second)) == len(ids_second) + new = next(c for c in second.clusters if c.cluster_id == "er_000009") + assert new.canonical_record_id == 7 + + +def _chain_report() -> EntityResolutionReport: + pairs = [_pair(1, 2, "match"), _pair(2, 3, "match")] + return _manual_report(pairs, [EntityCluster("er_000000", (1, 2, 3), 3, 3, 0.9)], 3) + + +def test_split_keeps_id_with_the_canonical_record(): + reject = ReviewDecision("reject", left_id=1, right_id=2) + out = apply_review_decisions(_chain_report(), [reject]) + assert [(c.cluster_id, c.record_ids, c.canonical_record_id) for c in out.clusters] == [ + ("er_000000", (2, 3), 3) + ] + + +def test_split_away_from_canonical_gets_fresh_id(): + reject = ReviewDecision("reject", left_id=2, right_id=3) + out = apply_review_decisions(_chain_report(), [reject]) + assert [(c.cluster_id, c.record_ids, c.canonical_record_id) for c in out.clusters] == [ + ("er_000003", (1, 2), 1) + ] + + +def test_accept_on_resolved_report_keeps_untouched_cluster_ids(): + # End to end: (1,2) cluster; (3,4) in review band. Accepting (3,4) must not + # renumber the existing cluster or change its canonical record. + df = pd.DataFrame( + { + "id": [1, 2, 3, 4, 5], + "name": ["alice smith", "alice smith", "bob jones", "robert jones", "carol lee"], + "email": ["a@x.com", "a@x.com", "bob@y.com", "bob@y.com", "carol@z.com"], + } + ) + cfg = EntityResolutionConfig( + enabled=True, + backend="pandas", + blocking_rules=(BlockingRule("lower(l.email) = lower(r.email)"),), + comparisons=( + ComparisonLevel("name", "jaro_winkler", threshold=0.99, weight=2.0), + ComparisonLevel("email", "exact", weight=1.0), + ), + match_threshold=0.95, + clerical_review_threshold=0.3, + ) + frame, report = resolve_entities(df, config=cfg) + assert [set(c.record_ids) for c in report.clusters] == [{1, 2}] + before = report.clusters[0] + + out = apply_review_decisions(report, [ReviewDecision("accept", left_id=3, right_id=4)]) + kept = next(c for c in out.clusters if set(c.record_ids) == {1, 2}) + assert kept.to_dict() == before.to_dict() + new = next(c for c in out.clusters if set(c.record_ids) == {3, 4}) + assert new.cluster_id not in set(frame["cluster_id"])