#!/usr/bin/env python3 """Checks a reviewed CRM merge plan before a person applies it. Read-only. Usage, from a working folder outside any repository, calling the script by its path: python3 "$SKILL_DIR/scripts/check_merge_plan.py" PLAN.csv --crm hubspot|salesforce|attio|generic \ [--profile out/profile.json] [--draft] The plan uses the columns of assets/merge-plan-template.csv. The script prints BLOCK, WARN and NOTE lines plus the merge sequence per group, and writes nothing. --profile is the profile.json written by profile_crm_export.py; with it, records missing from the profile, merges the profile did not group, unique keys the plan left blank and converted Salesforce leads are checked too. --draft is for the agent's hand-over: a missing approval or consent review prints PENDING instead of BLOCK, because only the named owners fill those fields. Exit code: 0 no BLOCK, 1 at least one BLOCK, 2 unreadable file or missing columns. Python 3.9 or later, standard library only. """ import argparse import csv import json import math import sys from collections import OrderedDict COLUMNS = ["group_id", "object", "record_id", "role", "decision", "band", "evidence", "unique_key_value", "conflicts_resolved", "opted_out", "bounced", "consent_reviewed", "external_ids_to_update", "approver", "approved_at", "executed_at", "result_record_id", "notes"] DECISIONS = {"merge", "keep_separate", "investigate"} ROLES = {"survivor", "merge", "keep"} YES = {"yes", "y", "true", "1"} ATTIO_MERGEABLE = {"contacts", "people", "person", "persons", "companies", "company"} SF_NO_MERGE = {"deals", "deal", "opportunities", "opportunity"} def yes(v): return (v or "").strip().lower() in YES def load(path): try: with open(path, newline="", encoding="utf-8-sig") as fh: reader = csv.DictReader(fh) header = reader.fieldnames or [] rows = list(reader) except (OSError, csv.Error, UnicodeDecodeError) as e: print(f"Cannot read {path}: {e}", file=sys.stderr) sys.exit(2) missing = [c for c in COLUMNS if c not in header] if not rows or missing: print(f"{path}: " + (f"missing columns {', '.join(missing)}" if missing else "has a header row and no plan rows"), file=sys.stderr) sys.exit(2) return [{k: (v or "").strip() for k, v in r.items()} for r in rows] MECHANICS = { "hubspot": "HubSpot merges two records of one object at a time and returns a new record ID unless the account is in the " "primary ID preservation beta; read each result back into result_record_id. Merges cannot be undone.", "salesforce": "Salesforce merges up to three records per operation; the survivor keeps its ID and losing records go to the " "Recycle Bin with MasterRecordId pointing at it. Created By and Created Date come from the oldest record.", "attio": "Attio's UI merges one pair of people or companies at a time; each merge creates a new record ID and URL, the " "record on the right wins conflicts, and deleted records cannot be recovered.", "generic": "Use the CRM's own merge tool and record the resulting ID for every merge.", } def sequence(crm, survivor, losers): if crm == "salesforce": ops = [losers[i:i + 2] for i in range(0, len(losers), 2)] return "; ".join(f"operation {n}: merge {', '.join(o)} into {survivor}" for n, o in enumerate(ops, start=1)) steps, target = [], survivor for n, loser in enumerate(losers, start=1): steps.append(f"step {n}: merge {loser} into {target}") target = f"the ID read back after step {n}" return "; ".join(steps) def main(argv=None): p = argparse.ArgumentParser(description="Read-only check of a CRM merge plan.") p.add_argument("plan") p.add_argument("--crm", choices=["hubspot", "salesforce", "attio", "generic"], required=True) p.add_argument("--profile") p.add_argument("--draft", action="store_true", help="print missing approvals and consent reviews as PENDING (hand-over check before sign-off)") a = p.parse_args(argv) rows = load(a.plan) profile = None if a.profile: try: with open(a.profile, encoding="utf-8") as fh: profile = json.load(fh) except (OSError, ValueError) as e: print(f"Cannot read --profile: {e}", file=sys.stderr) return 2 out = {"BLOCK": [], "WARN": [], "NOTE": [], "PENDING": []} signoff = "PENDING" if a.draft else "BLOCK" groups = OrderedDict() for r in rows: groups.setdefault(r["group_id"], []).append(r) seen = {} for r in rows: key = (r["object"].lower(), r["record_id"]) if key in seen and seen[key] != r["group_id"]: out["BLOCK"].append(f"{r['record_id']} ({r['object']}) is in groups {seen[key]} and {r['group_id']}: groups must be disjoint.") seen.setdefault(key, r["group_id"]) prof_ids, prof_group, prof_keys, prof_conv = {}, {}, {}, set() if profile: prof_ids = {o: set(ids) for o, ids in profile.get("record_ids", {}).items()} for g in profile.get("duplicates", {}).get("groups", []): prof_group[(g["object"], g["record_id"])] = g["group_id"] prof_keys = profile.get("unique_keys", {}) prof_conv = set(profile.get("converted_leads", [])) seq = [] for gid, rs in groups.items(): bad = [r["record_id"] for r in rs if r["decision"] not in DECISIONS or r["role"] not in ROLES] if bad: out["BLOCK"].append(f"{gid}: decision must be merge, keep_separate or investigate and role survivor, merge or keep " f"({', '.join(bad)}).") continue decisions, objects = {r["decision"] for r in rs}, {r["object"].lower() for r in rs} if len(decisions) > 1 or len(objects) > 1: out["BLOCK"].append(f"{gid}: rows carry different decisions or objects; split the group.") continue obj = objects.pop() if profile: for r in rs: if obj in prof_ids and r["record_id"] not in prof_ids[obj]: out["WARN"].append(f"{gid}: {r['record_id']} is not in the profile's {obj} export; re-check the ID.") if decisions.pop() != "merge": continue surv = [r for r in rs if r["role"] == "survivor"] losers = [r for r in rs if r["role"] == "merge"] if len(surv) != 1 or not losers or any(r["role"] == "keep" for r in rs): out["BLOCK"].append(f"{gid}: a merge group needs exactly one survivor, at least one merge row and no keep rows " f"(found {len(surv)} survivors, {len(losers)} merge rows).") continue if any(r["band"].lower() == "possible" for r in rs): out["BLOCK"].append(f"{gid}: possible band; after review, re-band to probable and write the new evidence in the " "evidence column.") if obj in ("leads", "lead"): conv = [r["record_id"] for r in rs if r["record_id"] in prof_conv] if conv: out["BLOCK"].append(f"{gid}: {', '.join(conv)} is a converted lead; never merge it: convert the open lead into " "the existing contact instead (D07).") elif not profile and a.crm == "salesforce": out["NOTE"].append(f"{gid}: pass --profile so converted leads in lead merges are blocked.") key_of = {r["record_id"]: ", ".join(sorted({k for k in (r["unique_key_value"], prof_keys.get(obj, {}).get(r["record_id"], "")) if k})) for r in rs} keys = {k for r in rs for k in (r["unique_key_value"], prof_keys.get(obj, {}).get(r["record_id"], "")) if k} if len(keys) > 1: out["BLOCK"].append(f"{gid}: different unique keys ({', '.join(sorted(keys))}); these are different source records.") flagged = [r["record_id"] for r in rs if yes(r["opted_out"]) or yes(r["bounced"])] if flagged and not all(yes(r["consent_reviewed"]) for r in rs): out[signoff].append(f"{gid}: {', '.join(flagged)} opted out or bounced; a merge can widen permission, so the privacy " "owner reviews and consent_reviewed is yes on every row.") for r in rs: if not r["approver"] or not r["approved_at"]: out[signoff].append(f"{gid}: {r['record_id']} has no approver or approved_at.") for r in losers: if key_of[r["record_id"]] and not r["external_ids_to_update"]: out["BLOCK"].append(f"{gid}: losing record {r['record_id']} holds key {key_of[r['record_id']]}; list the systems " "that store its ID in external_ids_to_update.") if a.crm == "attio" and obj not in ATTIO_MERGEABLE: out["BLOCK"].append(f"{gid}: Attio's UI merges people and companies only; {obj} needs a different repair.") if a.crm == "salesforce" and obj in SF_NO_MERGE: out["BLOCK"].append(f"{gid}: Salesforce has no native opportunity merge; move contact roles and activities, then " "close the duplicate through the change list.") for r in rs: if r["executed_at"] and not r["result_record_id"]: out["WARN"].append(f"{gid}: {r['record_id']} is marked executed without result_record_id; read the ID back.") if profile: pg = {prof_group.get((obj, r["record_id"])) for r in rs} if None in pg or len(pg) > 1: out["WARN"].append(f"{gid}: the profile did not group these records as certain or probable; the evidence column " "must give the reason for merging.") n = len(rs) if a.crm == "salesforce" and n > 3: out["NOTE"].append(f"{gid}: {n} records need {math.ceil((n - 1) / 2)} merge operations (up to three records each).") elif a.crm in ("hubspot", "attio") and n > 2: out["NOTE"].append(f"{gid}: {n} records need {n - 1} sequential merges; each later merge uses the ID read back " "from the previous one.") seq.append(f"{gid} {obj}: " + sequence(a.crm, surv[0]["record_id"], [r["record_id"] for r in losers])) print(f"Merge plan check (read-only). Plan: {a.plan.replace(chr(92), '/').split('/')[-1]}. CRM: {a.crm}. Nothing was changed.") for level in ("BLOCK", "WARN", "NOTE", "PENDING"): for msg in out[level]: print(f"{level} {msg}") if seq: print("Sequence (apply only after every BLOCK is cleared, first batch by a person, then read back):") for line in seq: print(f" {line}") print(MECHANICS[a.crm]) code = 1 if out["BLOCK"] else 0 pending = f", {len(out['PENDING'])} PENDING" if a.draft else "" print(f"Totals: {len(out['BLOCK'])} BLOCK, {len(out['WARN'])} WARN, {len(out['NOTE'])} NOTE{pending}. Exit {code}.") if a.draft and out["PENDING"]: print("Hand the plan to the CRM owner (and the privacy owner for PENDING consent lines); rerun without --draft " "after they sign: it must print 0 BLOCK before anyone merges.") return code if __name__ == "__main__": sys.exit(main())