"""Duplicate engine: blocking keys, pair comparison, bands, the union guard and the D checks. Works on tables already in memory: no file access, no network calls. """ from collections import Counter, defaultdict from datetime import date from hygiene_core import ( OBJECTS, PERSONS, PLACEHOLDER_LOCALS, ROLE_LOCALS, archived, base_label, email_parts, email_variant, idsort, is_placeholder, jw, latin, name_sim, pct, phone_key, plural, soundex) # ---------- D: duplicate engine ---------- class UF: def __init__(self): self.p = {} def find(self, x): self.p.setdefault(x, x) while self.p[x] != x: self.p[x] = self.p[self.p[x]] x = self.p[x] return x def union(self, a, b): self.p[self.find(a)] = self.find(b) BAND = {"certain": 3, "probable": 2, "possible": 1} BLOCK_LABEL = {"ukey": "unique key", "email": "email", "variant": "email variant", "phone": "phone", "name": "initial and surname sound", "sorted": "full name", "domname": "domain and surname sound", "domain": "domain", "name6": "name start and country", "company": "company and pipeline"} def ukey(d): """The record's unique key, or "" when it is blank or a placeholder.""" k = d.get("ukey", "") return k if k and not is_placeholder(k, "__ukey", set()) else "" def person_blocks(ctx, d): keys = [] if ukey(d): keys.append(("ukey", d["ukey"])) for e in d["emails"]: local, dom = email_parts(e) if local and local not in ROLE_LOCALS and local not in PLACEHOLDER_LOCALS: keys.append(("email", e)) keys.append(("variant", email_variant(e, ctx.dot_domains))) for k in d["pkeys"]: keys.append(("phone", k)) if d["name"]: first, sx = d["name"].split()[0][0], soundex(d["last_name"]) keys.append(("name", f"{first}|{sx}")) keys.append(("sorted", " ".join(sorted(d["name"].split())))) if d["reg"] and not d["personal"] and not d["role"]: keys.append(("domname", f"{d['reg']}|{sx}")) return keys def company_blocks(ctx, d): keys = [] if ukey(d): keys.append(("ukey", d["ukey"])) for dom in sorted(d["domains"]): if dom not in ctx.freemail: keys.append(("domain", dom)) if d["name"]: keys.append(("name6", f"{d['name'][:6]}|{d['country']}")) if d["pkey"]: keys.append(("phone", d["pkey"])) return keys def deal_blocks(ctx, d): keys = [] if ukey(d): keys.append(("ukey", d["ukey"])) for c in sorted(ctx.linked("deals", "companies", d["rid"])): keys.append(("company", f"{c}|{d['pipeline']}")) return keys def ukey_conflict(a, b): return bool(ukey(a) and ukey(b) and ukey(a) != ukey(b)) def same_ukey(a, b): return bool(ukey(a)) and ukey(a) == ukey(b) # Each compare function returns ([(rule_id, band, basis)], score). The basis names the evidence, so # calibration can tighten the one rule behind false matches (references/matching-rules.md). def compare_persons(ctx, obj, a, b, shared_phones): th, rules = ctx.th, [] usable = lambda d: {e for e in d["emails"] if email_parts(e)[0] and email_parts(e)[0] not in ROLE_LOCALS and email_parts(e)[0] not in PLACEHOLDER_LOCALS} ea, eb = usable(a), usable(b) if same_ukey(a, b): rules.append(("D01", "certain", "same unique key")) if ea & eb: rules.append(("D02", "certain", "same email")) elif {email_variant(e, ctx.dot_domains) for e in ea} & {email_variant(e, ctx.dot_domains) for e in eb}: rules.append(("D03", th["d03_band"], "plus tag or dots removed")) sim = name_sim(a["name"], b["name"]) dom = a["reg"] and a["reg"] == b["reg"] and not a["personal"] and not (a["role"] or b["role"]) comp = ctx.linked(obj, "companies", a["rid"]) & ctx.linked(obj, "companies", b["rid"]) phone = (set(a["pkeys"]) & set(b["pkeys"])) - shared_phones if sim >= th["name_similarity_probable"] and (dom or comp or phone): why = [w for w, hit in (("email domain", dom), ("company", comp), ("phone", phone)) if hit] rules.append(("D04", th["d04_band"], "similar name, same " + " and ".join(why))) elif sim >= th["name_similarity_possible"] and a["company_name"] and a["company_name"] == b["company_name"]: rules.append(("D04", "possible", "similar name, same company name")) return rules, max(sim, 1.0 if rules and rules[0][1] == "certain" else 0.0) def compare_companies(ctx, a, b): th, rules = ctx.th, [] if same_ukey(a, b): rules.append(("D01", "certain", "same unique key")) sim = jw(a["name"], b["name"]) if a["name"] and b["name"] and latin(a["name"] + b["name"]) else \ (1.0 if a["name"] and a["name"] == b["name"] else 0.0) own_a, own_b = a["domains"] - ctx.freemail, b["domains"] - ctx.freemail shared = own_a & own_b if shared: if sim >= th["company_similarity_probable"]: rules.append(("D05", "probable", "same domain, similar name")) else: rules.append(("D05", "possible", "same domain, different names")) same_name = a["name"] and a["name"] == b["name"] if same_name and a["pkey"] and a["pkey"] == b["pkey"]: rules.append(("D05", th["d05_name_phone_band"], "same name and phone")) elif same_name and a["country"] and a["country"] == b["country"]: if own_a and own_b and not shared: rules.append(("D05", "possible", "same name, different domains")) else: rules.append(("D05", th["d05_name_country_band"], "same name and country")) elif not shared and (not a["domain"] or not b["domain"]) and sim >= th["company_similarity_probable"] \ and a["country"] and a["country"] == b["country"]: rules.append(("D05", "possible", "similar name, same country, a domain missing")) return rules, max(sim, 1.0 if rules and rules[0][1] == "certain" else 0.0) def compare_deals(ctx, a, b): th, rules = ctx.th, [] if same_ukey(a, b): rules.append(("D01", "certain", "same unique key")) if "renewal" in a["deal_type"] or "renewal" in b["deal_type"]: return rules, 1.0 if rules else 0.0 if a["close"] and b["close"] and abs((a["close"] - b["close"]).days) >= th["renewal_close_gap_days"]: return rules, 1.0 if rules else 0.0 comp = ctx.linked("deals", "companies", a["rid"]) & ctx.linked("deals", "companies", b["rid"]) if comp and a["open"] and b["open"]: amt = isinstance(a["amount"], float) and isinstance(b["amount"], float) and \ abs(a["amount"] - b["amount"]) <= th["deal_amount_tolerance"] * max(abs(a["amount"]), abs(b["amount"])) near = a["created"] and b["created"] and abs((a["created"] - b["created"]).days) <= th["deal_created_window_days"] if a["pipeline"] == b["pipeline"] and amt and near: rules.append(("D06", th["d06_band"], "same company, pipeline and amount, created close together")) elif a["name"] and a["name"] == b["name"]: rules.append(("D06", "possible", "same open deal name")) return rules, 1.0 if rules else 0.0 def assoc_count(ctx, obj, d): others = {"contacts": ("companies", "deals"), "leads": ("companies",), "companies": ("contacts", "deals"), "deals": ("companies", "contacts")}[obj] return sum(len(ctx.linked(obj, o, d["rid"])) for o in others) def join_links(links, keys): """Unions certain and probable links, strongest first. A union that would put two different unique keys in one group is refused: that link becomes a possible pair with a key conflict. keys[i] is record i's unique key or "". Returns the union-find.""" uf, held = UF(), {} for l in sorted((l for l in links if l["band"] != "possible"), key=lambda l: (-BAND[l["band"]], -l["score"], l["i"], l["j"])): ra, rb = uf.find(l["i"]), uf.find(l["j"]) if ra == rb: continue ka = held.get(ra, {keys[l["i"]]} - {""}) kb = held.get(rb, {keys[l["j"]]} - {""}) if ka and kb and ka != kb: l.update(band="possible", conflict=True, refused=True) continue uf.union(ra, rb) held[uf.find(rb)] = ka | kb return uf def run_duplicates(ctx): th, out = ctx.th, {} for obj in OBJECTS: if obj not in ctx.t: continue # Converted Salesforce leads are archived; an open lead matching one's contact is D07, not a merge. recs = [d for d in ctx.info[obj] if not d["test"] and d["rid"] and not archived(obj, d)] blockfn = person_blocks if obj in PERSONS else company_blocks if obj == "companies" else deal_blocks blocks, shared, shared_phones = defaultdict(set), [], set() for i, d in enumerate(recs): for k in blockfn(ctx, d): blocks[k].add(i) if obj in PERSONS: raw = defaultdict(set) for i, d in enumerate(recs): for p in d["phones"]: k = phone_key(p, allow_placeholder=True) if k: raw[k].add(i) for k, members in sorted(raw.items()): names = {recs[i]["name"] for i in members if recs[i]["name"]} if len(names) >= th["shared_phone_distinct_names"] or len(members) > th["oversized_block"]: shared_phones.add(k) shared.append({"kind": "phone", "value": "phone ending " + k[-2:], "records": len(members), "names": len(names), "record_ids": sorted((recs[i]["rid"] for i in members), key=idsort)}) cand = set() for k, members in blocks.items(): if len(members) < 2 or (k[0] == "phone" and k[1] in shared_phones): continue if len(members) > th["oversized_block"]: keyed = k[0] in ("ukey", "email") shared.append({"kind": k[0], "value": "block too large to compare", "records": len(members), "names": 0, "record_ids": sorted((recs[i]["rid"] for i in members), key=idsort) if keyed else []}) continue m = sorted(members) cand.update((m[x], m[y]) for x in range(len(m)) for y in range(x + 1, len(m))) links = [] for i, j in sorted(cand): a, b = recs[i], recs[j] if obj in PERSONS: rules, score = compare_persons(ctx, obj, a, b, shared_phones) elif obj == "companies": rules, score = compare_companies(ctx, a, b) else: rules, score = compare_deals(ctx, a, b) if not rules: continue band = max((r[1] for r in rules), key=BAND.get) basis = "; ".join(sorted({f"{r[0]} {r[2]}" for r in rules if r[1] == band})) conflict = ukey_conflict(a, b) if conflict: band = "possible" links.append({"i": i, "j": j, "rules": sorted({r[0] for r in rules}), "band": band, "basis": basis, "score": round(score, 3), "conflict": conflict, "refused": False}) keys = [ukey(d) for d in recs] uf = join_links(links, keys) members = defaultdict(set) for l in links: if l["band"] != "possible": members[uf.find(l["i"])].update((l["i"], l["j"])) groups = [] for root, mem in members.items(): inner = [l for l in links if l["band"] != "possible" and l["i"] in mem] ordered = sorted(mem, key=lambda i: ( -1 if keys[i] else 0, -assoc_count(ctx, obj, recs[i]), 0 if ("active" in ctx.owner_state(recs[i]["owner"]) and "integration" not in ctx.owner_state(recs[i]["owner"])) else 1, recs[i]["created"] or date.max, idsort(recs[i]["rid"]))) first, second = recs[ordered[0]], recs[ordered[1]] if bool(keys[ordered[0]]) != bool(keys[ordered[1]]): why = "holds the unique key" elif assoc_count(ctx, obj, first) != assoc_count(ctx, obj, second): why = f"more associations ({assoc_count(ctx, obj, first)} vs {assoc_count(ctx, obj, second)})" elif ctx.owner_state(first["owner"]) != ctx.owner_state(second["owner"]) and \ "active" in ctx.owner_state(first["owner"]) and "integration" not in ctx.owner_state(first["owner"]): why = "owner is an active person" elif (first["created"] or date.max) != (second["created"] or date.max): why = "oldest create date" else: why = "tie on every criterion: decide in review" groups.append({"object": obj, "band": "certain" if all(l["band"] == "certain" for l in inner) else "probable", "members": [recs[i] for i in ordered], "rules": sorted({r for l in inner for r in l["rules"]}), "score": max(l["score"] for l in inner), "conflict": len({keys[i] for i in mem} - {""}) > 1, "why": why, "links": inner}) groups.sort(key=lambda g: (-BAND[g["band"]], idsort(g["members"][0]["rid"]))) pairs = [l for l in links if l["band"] == "possible"] out[obj] = {"recs": recs, "links": links, "groups": groups, "pairs": pairs, "shared": shared} cross = [] if ctx.crm == "salesforce" and "leads" in out and "contacts" in out: by_mail = defaultdict(list) for d in out["contacts"]["recs"]: if d["email"] and not d["role"]: by_mail[d["email"]].append(d) for d in out["leads"]["recs"]: if d["email"] and not d["role"] and not d["converted"]: for c in by_mail.get(d["email"], []): cross.append({"lead": d, "contact": c}) out["_cross"] = cross return out def checks_d(ctx, res, dup): th = ctx.th for obj in OBJECTS: if obj not in dup: continue o, recs = dup[obj], dup[obj]["recs"] n = len(recs) def emit(cid, oversized=()): ls = [l for l in o["links"] if cid in l["rules"]] ids = {recs[k]["rid"] for l in ls for k in (l["i"], l["j"])} gs = [g for g in o["groups"] if cid in g["rules"]] ps = [l for l in o["pairs"] if cid in l["rules"]] parts = [plural(sum(1 for g in gs if g["band"] == b), f"{b} group") for b in ("certain", "probable")] parts.append(plural(len(ps), "possible pair") + f" ({sum(1 for p in ps if p['conflict'])} with a key conflict)") certain = any(l["band"] == "certain" for l in ls) st = ("FAIL" if certain else "WARN") if cid in ("D01", "D02") and ls else "WARN" if ls else "PASS" detail = ("Rule fired in " + ", ".join(parts) + ".") if ls else "No candidate pairs." for s in oversized: st = "WARN" if st == "PASS" else st detail += f" One value on {s['records']} records was too large to compare: not unique (see D08)." ids |= set(s["record_ids"]) res.add(cid, obj, st, len(ids), n, base_label(obj), "FAIL at 1 certain group" if cid in ("D01", "D02") else "any", detail, ids) return ls if "__ukey" in ctx.t[obj].col: emit("D01", [s for s in o["shared"] if s["kind"] == "ukey"]) else: res.nc("D01", obj, f"no unique key declared (--unique-key {obj}=Header)") if obj in PERSONS: ls = [l for l in o["links"] if "D02" in l["rules"]] roles = defaultdict(set) for d in recs: if d["role"] and d["email"]: roles[d["email"]].add(d["rid"]) shared_roles = {e: r for e, r in roles.items() if len(r) > 1} big_mail = [s for s in o["shared"] if s["kind"] == "email"] if ls or big_mail: emit("D02", big_mail) elif shared_roles: res.add("D02", obj, "INFO", sum(len(r) for r in shared_roles.values()), n, base_label(obj), "", f"Shared role mailbox: {plural(len(shared_roles), 'address')} on several records; never grouped.", {i for r in shared_roles.values() for i in r}) else: emit("D02") emit("D03") emit("D04") if obj == "companies": emit("D05") if obj == "deals": if ctx.has_assoc_data("deals", "companies"): emit("D06") else: res.nc("D06", obj, "needs deal to company associations (export with associations)") if obj == "leads": if ctx.crm != "salesforce": res.na("D07", obj, "lead conversion matching applies to Salesforce leads") elif "contacts" not in dup: res.nc("D07", obj, "needs the contacts export") else: cross = dup["_cross"] per = {c["lead"]["rid"]: f"matches contact {c['contact']['rid']}; convert into it" for c in cross} res.add("D07", obj, "WARN" if per else "PASS", len(per), n, "unconverted leads", "any", "", per, per=per) phones = [s for s in o["shared"] if s["kind"] == "phone"] keyed = [s for s in o["shared"] if s["kind"] in ("ukey", "email")] big = Counter(BLOCK_LABEL.get(s["kind"], s["kind"]) for s in o["shared"] if s["kind"] not in ("phone", "ukey", "email")) notes8 = [f"{s['value']} on {s['records']} records ({s['names']} names)" for s in phones[:5]] if len(phones) > 5: notes8.append(f"{len(phones) - 5} more shared phones") notes8 += [f"one {BLOCK_LABEL[s['kind']]} value on {s['records']} records: not unique, fix the key or the writer" for s in keyed] if big: notes8.append("blocks too large to compare: " + ", ".join(f"{k} {v}" for k, v in sorted(big.items()))) res.add("D08", obj, "WARN" if keyed else "INFO" if o["shared"] else "PASS", len(o["shared"]), None, "shared or oversized blocking value(s)", "", "; ".join(notes8) + (". Never paired." if o["shared"] else ""), {i for s in o["shared"] for i in s["record_ids"]}) notes, chain_ids = [], set() for g in o["groups"]: if len(g["members"]) > th["max_group_size"]: notes.append(f"group of {len(g['members'])} records: split by hand and check the key") chain_ids.update(m["rid"] for m in g["members"]) for rule in sorted({r for l in o["links"] for r in l["rules"]}): ids = {recs[k]["rid"] for l in o["links"] if rule in l["rules"] for k in (l["i"], l["j"])} sizes = [len(g["members"]) for g in o["groups"] if rule in g["rules"]] + \ [2 for l in o["pairs"] if rule in l["rules"]] if n and len(ids) / n > th["rule_breadth_share"] and sizes and sum(sizes) / len(sizes) > th["rule_breadth_avg_group"]: notes.append(f"{rule} flags {pct(len(ids), n)} of records with an average group of {sum(sizes) / len(sizes):.1f}: tighten it") res.add("D09", obj, "WARN" if notes else "PASS", len(chain_ids), n, base_label(obj), f"groups above {th['max_group_size']}; rules above {th['rule_breadth_share']:.0%} with groups above {th['rule_breadth_avg_group']}", "; ".join(notes), chain_ids)