#!/usr/bin/env python3 """Read-only profiler for CRM CSV exports: HubSpot, Salesforce, Attio or any CRM. Usage, from a working folder outside any repository, calling the script by its path: python3 "$SKILL_DIR/scripts/profile_crm_export.py" --crm hubspot \ --contacts exports/contacts.csv --companies exports/companies.csv --deals exports/deals.csv \ --users exports/users.csv --rules crm-hygiene-rules.json \ --unique-key "contacts=Product User ID" --expected-count contacts=1250 \ --as-of 2026-11-01 --calibration 20 --out out python3 "$SKILL_DIR/scripts/profile_crm_export.py" --self-test Every object file is optional, but at least one of --contacts, --companies, --deals or --leads is needed. Cross-object checks run only when both files are present. The script maps columns, checks export integrity and coverage, runs the X, F, E, R, D, O, L and A checks with numerators and denominators, builds the duplicate review queue and writes local files under --out: summary.md, findings.csv, duplicates.csv, possible-pairs.csv, columns.csv, profile.json, trend.md and calibration-sample.csv. The code is split across this file and the hygiene_*.py modules beside it, so keep the scripts folder together. It never calls a network service, never writes outside --out (no bytecode cache either) and prints counts, statuses and record IDs only. Names, emails and phone numbers are masked in its files unless --unmask is given. Exit code: 0 no FAIL, 1 at least one FAIL, 2 unreadable input, a file with no records, no record ID column or invalid arguments. Python 3.9 or later, standard library only. """ import os import sys sys.dont_write_bytecode = True # keep the skill folder free of __pycache__ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # python3 -I leaves the script folder off the path import argparse import hashlib import json from collections import defaultdict from datetime import datetime, timezone from hygiene_core import ( InputError, OBJECTS, STATUSES, TYPO_DOMAINS, as_amount, bounce_class, company_norm, country_code, email_valid, email_variant, is_placeholder, jw, mask_owner, name_sim, norm_domain, norm_email, person_name, phone_key, phone_placeholder, registrable, sf_id_damage, sf_suffix, sf_valid18, soundex) from hygiene_io import Table, header_index, map_columns, read_table from hygiene_checks import Ctx, Results, checks_a, checks_e_r, checks_f, checks_l, checks_o, checks_x from hygiene_duplicates import checks_d, compare_companies, compare_persons, join_links, person_blocks, run_duplicates from hygiene_outputs import coverage_line, line, write_outputs DEFAULT_RULES = { "version": "unversioned", "b2b": True, "internal_domains": [], "test_domains": [], "freemail_domains_add": [], "dot_insensitive_domains": ["gmail.com", "googlemail.com"], "typo_domains": {}, "integration_owner_patterns": ["integration", "api", "sync", "system", "automation"], "lifecycle_order": ["subscriber", "lead", "marketingqualifiedlead", "salesqualifiedlead", "opportunity", "customer", "evangelist", "other"], "owner_required_lifecycle": ["salesqualifiedlead", "opportunity", "customer"], "owner_critical_customer_objects": ["contacts", "companies"], "contact_owner_from_company": False, "customer_lifecycle_values": ["customer"], "customer_evidence_field": "", "closed_won_stages": ["closedwon"], "closed_lost_stages": ["closedlost"], "open_lead_status_values": ["new", "open", "inprogress", "attemptedtocontact", "connected"], "pipelines": {}, "required": {"contacts": [], "companies": [], "deals": [], "leads": []}, "allowed_values": {}, "unique_keys": {}, "zero_amount_deal_types": [], "deal_company_required": False, "lost_reason_required": True, "exemptions": {}, "thresholds": { "stale_open_deal_days_past_close": 30, "stale_open_deal_fail_share": 0.10, "no_activity_days": 60, "dormant_days": 365, "dormant_warn_share": 0.10, "never_engaged_days": 180, "non_critical_missing_fail_share": 0.20, "non_critical_missing_fail_min_records": 20, "integration_owner_fail_share": 0.05, "customer_without_evidence_fail_share": 0.05, "name_similarity_probable": 0.92, "name_similarity_possible": 0.88, "company_similarity_probable": 0.93, "deal_amount_tolerance": 0.05, "deal_created_window_days": 30, "renewal_close_gap_days": 300, "oversized_block": 50, "shared_phone_distinct_names": 4, "max_group_size": 10, "rule_breadth_share": 0.10, "rule_breadth_avg_group": 3, "queue_max_days": 2, "d03_band": "probable", "d04_band": "probable", "d05_name_country_band": "probable", "d05_name_phone_band": "probable", "d06_band": "probable", }, } BAND_KEYS = ("d03_band", "d04_band", "d05_name_country_band", "d05_name_phone_band", "d06_band") # ---------- self-test ---------- def run_on_rows(check, crm, rules_changes, **rows): """Runs one check family on small tables built in memory and returns its results. rows: object=(headers, row, ...).""" tables = {} for obj, (headers, *cells) in rows.items(): tables[obj] = Table(obj, obj + ".csv") tables[obj].headers, tables[obj].rows = headers, [list(c) for c in cells] map_columns(tables[obj], crm, {}) args = argparse.Namespace(crm=crm, as_of_date=datetime(2026, 11, 1).date(), date_order=None, allow_reserved_domains=True) res = Results({}) check(Ctx(args, dict(json.loads(json.dumps(DEFAULT_RULES)), **rules_changes), tables), res) return res.checks def self_test(): at, ok = "@", 0 cases = [ (jw("martha", "martha") == 1.0, "identical strings score 1.0"), (abs(jw("martha", "marhta") - 0.961) < 0.001, "martha and marhta score about 0.961"), (email_valid(norm_email("jane.doe" + at + "example.org")), "valid address accepted"), (not email_valid("a" + at + "b" + at + "example.org"), "two at signs rejected"), (not email_valid("x" * 65 + at + "example.org"), "65-octet local part rejected"), (not email_valid("jane" + at + "example.o"), "one-letter top label rejected"), (phone_key("12345") == "", "fewer than 7 digits invalid"), (phone_key("1" * 16) == "", "more than 15 digits invalid"), (phone_placeholder("0000000000") and phone_key("0000000000") == "", "all-same-digit phone is a placeholder"), (registrable(norm_domain("https://www.shop.example.com:8080/path?q=1")) == "example.com", "URL to registrable domain"), (registrable("shop.example.co.uk") == "example.co.uk", "co.uk keeps three labels"), (sf_suffix("00300000003T2PG") == "AA0" and sf_valid18("00300000003T2PGAA0"), "Salesforce documented ID pair"), (not sf_valid18("00300000003t2pgaa0"), "lowercased 18-character ID fails"), (bounce_class("550 5.1.1 user unknown") == "address-dead", "5.1.1 is address-dead"), (bounce_class("550 5.7.1 policy rejection") == "sender-side", "5.7.1 is sender-side"), (bounce_class("421 4.7.0 try later") == "temporary", "4.7.0 is temporary"), (bounce_class("550 5.2.2 mailbox full") == bounce_class("Mailbox full") == "mailbox-full", "a full mailbox is not decay"), (bounce_class("UNKNOWN_USER") == "address-dead", "enum-style bounce reasons are read"), (bounce_class("550 5.1.10 RESOLVER.ADR.RecipientNotFound; Recipient not found by SMTP address lookup") == "address-dead", "5.1.10 recipient not found is address-dead"), (bounce_class("Recipient not found by Microsoft Exchange") == bounce_class("Address not found") == "address-dead", "recipient not found and address not found are address-dead"), (bounce_class("Delivery failed at Microsoft Exchange") == "unknown", "'soft' inside 'Microsoft' is not a soft bounce"), (bounce_class("Rejected by security software") == "unknown" and bounce_class("SOFT_BOUNCE") == "temporary", "'software' is not a soft bounce; a whole-word soft is"), (email_valid("jane" + at + "example.xn--p1ai") and email_valid("jane" + at + "b\u00fccher.example"), "internationalised domains accepted"), (not email_valid("jane" + at + "a" * 64 + ".example"), "64-octet domain label rejected"), (email_variant("lee.wong+news" + at + "example.com", {"example.com"}) == "leewong" + at + "example.com", "variant key"), (TYPO_DOMAINS.get("gmial" + ".com") == "gmail.com" and email_valid("someone" + at + "gmial" + ".com"), "typo domain looks valid"), (company_norm("The Bluefin Logistics Ltd") == "bluefin logistics", "legal form and leading 'the' removed"), (country_code("U.S.A.") == "US" and country_code("Deutschland") == "DE", "country aliases"), (soundex("Smith") == soundex("Smyth") == "S530", "phonetic block for Smith and Smyth"), (name_sim("lee sam", "sam lee") == 1.0, "swapped first and last name"), (name_sim(person_name("", "", "\u738b\u4f1f"), person_name("", "", "\u738b\u4f1f")) == 1.0 and name_sim("\u738b\u4f1f", "\u738b\u4f1e") == 0.0, "non-Latin names use exact equality"), ] blank = {"ukey": "", "emails": [], "name": "", "last_name": "", "pkeys": [], "reg": "", "personal": False, "role": False} cases.append((person_blocks(type("C", (), {"dot_domains": set()})(), blank) == [], "blank records form no blocks")) cases.append((as_amount("1,234.56") == as_amount("1.234,56") == 1234.56 and as_amount("12,5") == 12.5 and as_amount("") is None and as_amount("1,23,456") == "ambiguous", "amount separators")) cases.append((mask_owner("Deactivated/Removed (casey" + at + "example.com)") == "Deactivated/Removed (c***" + at + "e***.com)", "emails inside owner labels are masked")) cases.append((is_placeholder("n/a", "title", set()) and is_placeholder("NA", "state", set()) and not is_placeholder("New", "lead_status", {"new"}), "placeholders, with allowed values exempt")) links = [{"i": 0, "j": 1, "band": "certain", "score": 1.0}, {"i": 1, "j": 2, "band": "probable", "score": 1.0}] uf = join_links(links, ["K1", "", "K2"]) cases.append((links[1]["band"] == "possible" and links[1].get("refused") and uf.find(0) == uf.find(1) != uf.find(2), "a record without a key never bridges two different keys into one group")) fake = type("C", (), {"th": DEFAULT_RULES["thresholds"], "freemail": set()})() co = lambda dom: {"ukey": "", "name": "summit consulting", "domains": {dom}, "domain": dom, "country": "US", "pkey": ""} cases.append((compare_companies(fake, co("summitconsulting.example"), co("summit-advisors.example"))[0] == [("D05", "possible", "same name, different domains")], "same company name, different domains is possible")) fake.th = dict(DEFAULT_RULES["thresholds"], d05_name_country_band="possible") cases.append((compare_companies(fake, co("summitconsulting.example"), dict(co(""), domains=set(), domain=""))[0] == [("D05", "possible", "same name and country")], "a band setting demotes its own rule")) pctx = type("C", (), {"th": dict(DEFAULT_RULES["thresholds"], d04_band="possible"), "dot_domains": set(), "linked": lambda self, *a: set()})() pa = {"rid": "1", "ukey": "", "emails": ["jo.lee" + at + "acme.example"], "name": "jo lee", "reg": "acme.example", "personal": False, "role": False, "pkeys": [], "company_name": ""} cases.append((compare_persons(pctx, "contacts", pa, dict(pa, rid="2", emails=["jlee" + at + "acme.example"]), set())[0] == [("D04", "possible", "similar name, same email domain")], "d04_band demotes identical names at one domain")) a15, b15 = "00Q5e00000S14aB", "00Q5e00000S14Ab" cases.append((sf_id_damage([a15 + sf_suffix(a15), b15 + sf_suffix(b15)]) == ([], [], []), "valid 18-character IDs that differ only in letter case are different records")) cases.append((sf_id_damage([a15, b15, "00300000003t2pgaa0"]) == (sorted([a15, b15]), ["00300000003t2pgaa0"], sorted([a15, b15])), "15-character IDs equal once lowercased collide; a lowercased 18-character ID has a bad suffix")) people = (["Record ID", "Contact owner", "Lifecycle Stage", "Associated Company IDs"], ["1", "", "Customer", "9"], ["2", "Sam Ortiz", "Lead", "9"]) firms = (["Record ID", "Company owner"], ["9", "Sam Ortiz"]) staff = (["Owner ID", "Name", "Email", "Active", "Kind"], ["5", "Sam Ortiz", "", "true", "person"]) o01 = lambda **changes: next(c["status"] for c in run_on_rows(checks_o, "hubspot", changes, contacts=people, companies=firms, users=staff) if c["id"] == "O01" and c["object"] == "contacts") cases.append((o01() == "FAIL", "a customer contact without an owner fails by default")) cases.append((o01(owner_critical_customer_objects=["companies"]) == "WARN", "owner_critical_customer_objects demotes it")) cases.append((o01(contact_owner_from_company=True) == "PASS", "contact_owner_from_company counts the company's active owner")) no_flag = (["Id", "Email"], ["00Q5e00000S13Ld", "lee" + at + "example.org"]) cases.append((next(c["status"] for c in run_on_rows(checks_l, "salesforce", {}, leads=no_flag, contacts=no_flag) if c["id"] == "L10") == "NOT CHECKED", "L10 is NOT CHECKED without IsConverted, never a PASS of 0 of 0")) empty_pop = Results({}) empty_pop.add("L03", "deals", "PASS", 0, 0, "closed-won deals") cases.append((empty_pop.checks[0]["status"] == "NOT CHECKED", "a check with an empty population is NOT CHECKED, not PASS")) lead_rows = (["Id", "Email", "EmailBouncedDate", "CreatedDate"], ["00Q5e00000S20Ab", "lee" + at + "example.org", "2026-10-01", "2026-01-01"]) cases.append((next(c["status"] for c in run_on_rows(checks_e_r, "salesforce", {}, leads=lead_rows) if c["id"] == "E06") == "FAIL", "a Salesforce lead with only EmailBouncedDate counts as bounced")) for passed, label in cases: if not passed: print(f"Self-test failed: {label}") return 1 ok += 1 print(f"Self-test: {ok} assertions passed.") return 0 # ---------- main ---------- def load_rules(path, args): rules = json.loads(json.dumps(DEFAULT_RULES)) unknown = [] if path: try: with open(path, encoding="utf-8") as fh: user = json.load(fh) except (OSError, ValueError) as e: raise InputError(f"cannot read --rules: {e}") for k, v in user.items(): if k.startswith("_"): continue if k not in rules: unknown.append(k) elif k == "thresholds": rules[k].update({tk: tv for tk, tv in v.items() if not tk.startswith("_")}) elif k == "required": rules[k].update({ok: ov for ok, ov in v.items() if not ok.startswith("_")}) else: rules[k] = {dk: dv for dk, dv in v.items() if not dk.startswith("_")} if isinstance(v, dict) else v rules["freemail_domains_add"] = list(rules["freemail_domains_add"]) + [d for d in args.freemail_domains.split(",") if d.strip()] rules["internal_domains"] = list(rules["internal_domains"]) + [d for d in args.internal_domains.split(",") if d.strip()] for k in BAND_KEYS: if rules["thresholds"].get(k) not in ("probable", "possible"): raise InputError(f"thresholds.{k} must be probable or possible") crit = rules["owner_critical_customer_objects"] if not isinstance(crit, list) or any(o not in ("contacts", "companies", "leads") for o in crit): raise InputError("owner_critical_customer_objects must be a list of contacts, companies or leads") if not isinstance(rules["contact_owner_from_company"], bool): raise InputError("contact_owner_from_company must be true or false") meta = {"version": str(rules.get("version", "unversioned")), "unknown": unknown, "hash": hashlib.sha256(json.dumps(rules, sort_keys=True).encode()).hexdigest()[:12]} return rules, meta def parse_pairs(items, flag, number=False): out = {} for it in items: if "=" not in it: raise InputError(f"{flag} expects OBJECT=VALUE, got '{it}'") k, v = it.split("=", 1) k = k.strip().replace("-", "_") if number: if not v.strip().isdigit(): raise InputError(f"{flag} {k} must be a whole number") v = int(v) out[k] = v if number else v.strip() return out def main(argv=None): p = argparse.ArgumentParser(description="Read-only CRM CSV export profiler. Nothing is ever written to a CRM.", epilog="Exit codes: 0 no FAIL, 1 at least one FAIL, 2 unreadable input, no record ID " "column or invalid arguments. Options in detail: references/profiler.md.") p.add_argument("--crm", choices=["hubspot", "salesforce", "attio", "generic"], default="generic", help="header presets and CRM-specific checks (default generic)") for o, h in (("contacts", "contacts or people export"), ("companies", "companies or accounts export"), ("deals", "deals or opportunities export"), ("leads", "Salesforce leads export"), ("users", "users file: ID, name, email, active, kind (person, integration or queue)"), ("contact-roles", "Salesforce OpportunityContactRole export (OpportunityId, ContactId)"), ("queues", "queues file: ID and name")): p.add_argument(f"--{o}", metavar="FILE", help=h) p.add_argument("--map", action="append", default=[], metavar="OBJECT.FIELD=Header", help="map an export header to a canonical field; repeatable; wins over presets") p.add_argument("--rules", metavar="FILE", help="rules JSON with the keys of assets/rules-template.json") p.add_argument("--unique-key", action="append", default=[], metavar="OBJECT=Header", help="declare a unique key column for D01; repeatable") p.add_argument("--expected-count", action="append", default=[], metavar="OBJECT=N", help="record count the CRM shows for the export's filter; enables X08 coverage") p.add_argument("--previous", metavar="FILE", help="last month's profile.json; trend.md compares with it") p.add_argument("--as-of", metavar="YYYY-MM-DD", help="date for every age and window (default today, UTC)") p.add_argument("--out", default="crm-hygiene-out", metavar="DIR", help="output folder, created if missing") p.add_argument("--calibration", type=int, default=20, metavar="N", help="pairs per band per object in calibration-sample.csv (default 20; 0 leaves the file alone)") p.add_argument("--allow-reserved-domains", action="store_true", help="keep example.com, example.net, example.org and .example, .test, .invalid, .localhost " "addresses in the populations (fixtures only)") p.add_argument("--freemail-domains", default="", metavar="LIST", help="comma list of extra personal mail domains") p.add_argument("--internal-domains", default="", metavar="LIST", help="comma list of internal email domains, excluded with test records") p.add_argument("--date-order", choices=["mdy", "dmy"], help="order of slash dates that cannot be told apart; without it those date checks are NOT CHECKED") p.add_argument("--unmask", action="store_true", help="add clear names and emails to duplicates.csv for local review") p.add_argument("--json", action="store_true", help="print totals and check lines as JSON instead of text") p.add_argument("--self-test", action="store_true", help="run the built-in assertions and exit") args = p.parse_args(argv) if args.self_test: return self_test() if args.calibration < 0: print("--calibration must be 0 or more", file=sys.stderr) return 2 try: args.as_of_date = datetime.strptime(args.as_of, "%Y-%m-%d").date() if args.as_of else datetime.now(timezone.utc).date() except ValueError: print("--as-of must be YYYY-MM-DD", file=sys.stderr) return 2 try: args.expected = parse_pairs(args.expected_count, "--expected-count", number=True) ukeys = parse_pairs(args.unique_key, "--unique-key") maps = defaultdict(dict) for it in args.map: k, v = parse_pairs([it], "--map").popitem() if "." not in k: raise InputError(f"--map expects OBJECT.FIELD=Header, got '{it}'") obj, field = k.split(".", 1) maps[obj][field] = v rules, meta = load_rules(args.rules, args) files = {o: getattr(args, o.replace("-", "_")) for o in ("contacts", "companies", "deals", "leads", "users", "contact_roles", "queues")} if not any(files[o] for o in OBJECTS): raise InputError("give at least one of --contacts, --companies, --deals or --leads") tables = {} for obj, path in files.items(): if not path: continue t = read_table(obj, path) map_columns(t, args.crm, maps.get(obj, {})) if obj in OBJECTS and "record_id" not in t.col: raise InputError(f"{t.name}: no record ID column found; pass --map {obj}.record_id=Header") header = ukeys.get(obj) or rules["unique_keys"].get(obj) if header and obj in OBJECTS: i = header_index(t, header) if i is None: raise InputError(f"unique key column '{header}' not found in {t.name}") t.col["__ukey"], t.mapinfo["__ukey"] = i, ("mapped", t.headers[i], "unique key") if obj in OBJECTS: labels = [e["field"] for e in rules["required"].get(obj, [])] + \ [k.split(".", 1)[1] for k in rules["allowed_values"] if k.split(".", 1)[0] == obj] for lb in labels: i = header_index(t, lb) if i is not None and lb not in t.col: t.extra[lb] = i tables[obj] = t ctx = Ctx(args, rules, tables) res = Results(rules.get("exemptions", {})) checks_x(ctx, res) checks_f(ctx, res) checks_e_r(ctx, res) dup = run_duplicates(ctx) checks_d(ctx, res, dup) checks_o(ctx, res) checks_l(ctx, res) checks_a(ctx, res) checks, totals = write_outputs(ctx, res, dup, args, tables, meta) except InputError as e: print(f"Input error: {e}", file=sys.stderr) return 2 code = 1 if totals.get("FAIL") else 0 if args.json: print(json.dumps({"as_of": ctx.as_of.isoformat(), "totals": dict(totals), "exit": code, "checks": [{k: c[k] for k in ("id", "object", "field", "status", "numerator", "denominator")} for c in checks]}, indent=1)) return code print(f"CRM hygiene profile (read-only). As of {ctx.as_of.isoformat()}. Nothing was changed.") for name, t in tables.items(): print(coverage_line(name, t, checks)) for c in checks: if c["status"] in ("FAIL", "WARN"): print(f"{c['status']} " + line(c, ids=5)) print("Totals: " + ", ".join(f"{totals.get(s, 0)} {s}" for s in STATUSES) + f". Exit {code}.") print(f"Files in {args.out}: summary.md, findings.csv, duplicates.csv, possible-pairs.csv, columns.csv, profile.json, " "trend.md" + (", calibration-sample.csv" if args.calibration else "") + ".") return code if __name__ == "__main__": sys.exit(main())