"""Shared constants, normalisers and value parsing for profile_crm_export.py. Standard library only. This module opens no files and makes no network calls. """ import re import unicodedata from collections import defaultdict SCHEMA = "crm-hygiene-profile/1" OBJECTS = ("contacts", "companies", "deals", "leads") PERSONS = ("contacts", "leads") STATUSES = ("FAIL", "WARN", "INFO", "PASS", "NOT CHECKED", "N/A") FREEMAIL = set("""gmail.com googlemail.com yahoo.com yahoo.co.uk yahoo.fr yahoo.de ymail.com rocketmail.com outlook.com hotmail.com hotmail.co.uk hotmail.fr live.com live.co.uk msn.com icloud.com me.com mac.com aol.com proton.me protonmail.com pm.me gmx.com gmx.de gmx.net web.de mail.com zoho.com zohomail.com yandex.com yandex.ru mail.ru qq.com 163.com 126.com fastmail.com hey.com tutanota.com comcast.net verizon.net att.net sbcglobal.net btinternet.com orange.fr free.fr laposte.net libero.it t-online.de rediffmail.com""".split()) ROLE_LOCALS = set("""info marketing sales support abuse noc security postmaster hostmaster usenet news webmaster www uucp ftp admin billing contact hello help office team enquiries inquiries accounts hr jobs careers noreply no-reply donotreply""".split()) PLACEHOLDER_LOCALS = {"none", "noemail", "no-email", "test", "na", "unknown", "asdf"} TYPO_DOMAINS = { "gmial.com": "gmail.com", "gmal.com": "gmail.com", "gmaill.com": "gmail.com", "gamil.com": "gmail.com", "gnail.com": "gmail.com", "gmail.co": "gmail.com", "gmail.con": "gmail.com", "hotmial.com": "hotmail.com", "hotmal.com": "hotmail.com", "hotmail.con": "hotmail.com", "yahooo.com": "yahoo.com", "yaho.com": "yahoo.com", "yahoo.con": "yahoo.com", "outlok.com": "outlook.com", "outlook.con": "outlook.com", "iclod.com": "icloud.com", "icloud.con": "icloud.com", "aol.con": "aol.com", } PLACEHOLDER_VALUES = {"n/a", "na", "none", "null", "unknown", "tbd", "test", "asdf", "x", "-", "--", "."} TEST_NAMES = {"test", "asdf", "dummy", "sample"} HONORIFICS = {"mr", "mrs", "ms", "miss", "dr", "prof"} NAME_SUFFIXES = {"jr", "sr", "ii", "iii", "iv"} LEGAL_FORMS = set("""inc incorporated corp corporation co company llc llp lp ltd limited plc gmbh ag kg sa sas sarl srl spa bv nv oy ab as aps pty pte kk""".split()) TWO_LEVEL = {"co.uk", "org.uk", "ac.uk", "gov.uk", "com.au", "net.au", "org.au", "co.nz", "co.jp", "com.br", "com.mx", "co.in", "co.za", "com.sg"} RESERVED_DOMAINS = {"example.com", "example.net", "example.org"} RESERVED_TLDS = {"example", "test", "invalid", "localhost"} FIVE_DIGIT_POSTAL = {"US", "DE", "FR", "IT", "ES"} # 5.1.10 is "recipient address has null MX" (RFC 7505); Exchange Online also returns it for a recipient not found. ADDRESS_DEAD_CODES = {"5.1.0", "5.1.1", "5.1.2", "5.1.3", "5.1.6", "5.1.10", "5.2.1"} ADDRESS_DEAD_WORDS = ("unknown user", "user unknown", "no such user", "does not exist", "invalid recipient", "mailbox disabled", "recipient not found", "address not found") # RFC 3463 calls X.2.2 a persistent transient failure: the person can free space, so it is not decay. MAILBOX_FULL_WORDS = ("mailbox full", "mailbox is full", "over quota", "quota exceeded") SENDER_WORDS = ("content", "spam", "policy", "blocked", "reputation", "dmarc", "spf", "dkim", "authentication") TEMP_WORDS = ("pending", "soft", "try later", "temporar", "deferred") REMOVED_RE = re.compile(r"^\s*deactivated\s*/\s*removed\b", re.I) SCI_RE = re.compile(r"^[+-]?\d+(?:\.\d+)?[eE][+-]?\d+$") DOT0_RE = re.compile(r"^\d+\.0+$") FORMULA_RE = re.compile("^[=@\uff1d\uff20]\\S|^[+\\-\uff0b\uff0d][A-Za-z(\\[=]|^[\t\r\n]") FORMULA_START = ("=", "+", "-", "@", "\t", "\r", "\n", "\uff1d", "\uff0b", "\uff0d", "\uff20") MOJIBAKE_RE = re.compile("[\u00c3\u00c2][\u0080-\u00bf]|\u00e2\u20ac|\ufffd") ZERO_WIDTH = dict.fromkeys(map(ord, "\u200b\u200c\u200d\u2060\ufeff"), None) NBSP = "\u00a0" SF_SUFFIX = "ABCDEFGHIJKLMNOPQRSTUVWXYZ012345" COUNTRY_TABLE = """US|USA|United States|United States of America|U.S.|U.S.A.;GB|GBR|United Kingdom|UK| Great Britain|U.K.;CA|CAN|Canada;AU|AUS|Australia;NZ|NZL|New Zealand;IE|IRL|Ireland;DE|DEU|Germany| Deutschland;FR|FRA|France;ES|ESP|Spain|Espana;IT|ITA|Italy|Italia;NL|NLD|Netherlands|The Netherlands| Holland;BE|BEL|Belgium;CH|CHE|Switzerland;AT|AUT|Austria;SE|SWE|Sweden;NO|NOR|Norway;DK|DNK|Denmark; FI|FIN|Finland;PL|POL|Poland;PT|PRT|Portugal;CZ|CZE|Czechia|Czech Republic;IN|IND|India;SG|SGP| Singapore;JP|JPN|Japan;CN|CHN|China;HK|HKG|Hong Kong;KR|KOR|South Korea|Republic of Korea;BR|BRA| Brazil|Brasil;MX|MEX|Mexico;AR|ARG|Argentina;ZA|ZAF|South Africa;AE|ARE|United Arab Emirates|UAE; IL|ISR|Israel;TR|TUR|Turkey|Turkiye;NG|NGA|Nigeria;KE|KEN|Kenya;EG|EGY|Egypt;SA|SAU|Saudi Arabia; PH|PHL|Philippines;ID|IDN|Indonesia;MY|MYS|Malaysia;TH|THA|Thailand;VN|VNM|Vietnam|Viet Nam; PK|PAK|Pakistan;CO|COL|Colombia;CL|CHL|Chile;PE|PER|Peru;RO|ROU|Romania;GR|GRC|Greece;HU|HUN| Hungary;UA|UKR|Ukraine""" US_STATES = """AL Alabama;AK Alaska;AZ Arizona;AR Arkansas;CA California;CO Colorado;CT Connecticut; DE Delaware;DC District of Columbia;FL Florida;GA Georgia;HI Hawaii;ID Idaho;IL Illinois;IN Indiana; IA Iowa;KS Kansas;KY Kentucky;LA Louisiana;ME Maine;MD Maryland;MA Massachusetts;MI Michigan; MN Minnesota;MS Mississippi;MO Missouri;MT Montana;NE Nebraska;NV Nevada;NH New Hampshire; NJ New Jersey;NM New Mexico;NY New York;NC North Carolina;ND North Dakota;OH Ohio;OK Oklahoma; OR Oregon;PA Pennsylvania;RI Rhode Island;SC South Carolina;SD South Dakota;TN Tennessee;TX Texas; UT Utah;VT Vermont;VA Virginia;WA Washington;WV West Virginia;WI Wisconsin;WY Wyoming""" # ---------- normalizers ---------- def normh(s): return re.sub(r"[^a-z0-9]", "", (s or "").lower()) def fold(s): """Match key: NFKD without marks, lowercase, '&' to 'and', punctuation removed.""" s = unicodedata.normalize("NFKD", s or "").translate(ZERO_WIDTH) s = "".join(c for c in s if not unicodedata.combining(c)).lower().replace("&", " and ") return " ".join(re.sub(r"[^\w\s]|_", " ", s).split()) def vfold(s): """Value folding for stages and picklists: 'Closed Won' equals 'closedwon'.""" return re.sub(r"[\W_]+", "", unicodedata.normalize("NFKD", s or "").lower()) def latin(s): return all(c.isascii() or not c.isalpha() or "LATIN" in unicodedata.name(c, "") for c in s) def person_name(first, last, full): raw = f"{first} {last}".strip() or full toks = [t for t in fold(raw).split() if t not in HONORIFICS and t not in NAME_SUFFIXES] return " ".join(toks) def company_norm(name): toks = fold(name).split() if len(toks) > 1 and toks[0] == "the": toks = toks[1:] while len(toks) > 1 and toks[-1] in LEGAL_FORMS: toks.pop() return " ".join(toks) def jaro(a, b): if a == b: return 1.0 la, lb = len(a), len(b) if not la or not lb: return 0.0 win = max(max(la, lb) // 2 - 1, 0) ma, mb, m = [False] * la, [False] * lb, 0 for i, ca in enumerate(a): for j in range(max(0, i - win), min(lb, i + win + 1)): if not mb[j] and b[j] == ca: ma[i] = mb[j] = True m += 1 break if not m: return 0.0 t, k = 0, 0 for i in range(la): if ma[i]: while not mb[k]: k += 1 t += a[i] != b[k] k += 1 return (m / la + m / lb + (m - t / 2) / m) / 3 def jw(a, b): j, n = jaro(a, b), 0 for x, y in zip(a[:4], b[:4]): if x != y: break n += 1 return j + n * 0.1 * (1 - j) def name_sim(a, b): if not a or not b: return 0.0 if not (latin(a) and latin(b)): return 1.0 if a == b else 0.0 return max(jw(a, b), jw(" ".join(sorted(a.split())), " ".join(sorted(b.split())))) def soundex(s): s = re.sub(r"[^a-z]", "", fold(s)) if not s: return "" codes = {} for letters, code in (("bfpv", "1"), ("cgjkqsxz", "2"), ("dt", "3"), ("l", "4"), ("mn", "5"), ("r", "6")): codes.update(dict.fromkeys(letters, code)) out, prev = s[0].upper(), codes.get(s[0], "") for c in s[1:]: code = codes.get(c, "") if code and code != prev: out += code if c not in "hw": prev = code return (out + "000")[:4] def norm_email(e): return (e or "").translate(ZERO_WIDTH).replace(NBSP, " ").strip().lower() def email_parts(e): if e.count("@") != 1: return "", "" local, domain = e.split("@") return local, domain def email_valid(e): local, domain = email_parts(e) if not local or not domain or len(e) > 254 or len(local.encode("utf-8")) > 64: return False if " " in e or ".." in e or local[0] == "." or local[-1] == ".": return False if not domain.isascii(): try: domain = domain.encode("idna").decode("ascii") # internationalised domain to its xn-- form except UnicodeError: return False labels = domain.split(".") if len(domain) > 255 or len(labels) < 2 or any(not lb or len(lb) > 63 or lb[0] == "-" or lb[-1] == "-" for lb in labels): return False if any(not re.fullmatch(r"[a-z0-9-]+", lb) for lb in labels): return False return bool(re.fullmatch(r"[a-z]{2,}|xn--[a-z0-9-]+", labels[-1])) def email_variant(e, dot_domains): local, domain = email_parts(e) if not local: return "" local = local.split("+", 1)[0] if domain in dot_domains: local = local.replace(".", "") return local + "@" + domain def norm_domain(v): v = (v or "").translate(ZERO_WIDTH).strip().lower() if not v: return "" if "@" in v and "://" not in v: v = v.rsplit("@", 1)[1] v = re.sub(r"^[a-z][a-z0-9+.-]*://", "", v) v = re.split(r"[/?#]", v, 1)[0].rsplit("@", 1)[-1].split(":", 1)[0].strip(".") return v[4:] if v.startswith("www.") else v def registrable(d): parts = d.split(".") if len(parts) >= 3 and ".".join(parts[-2:]) in TWO_LEVEL: return ".".join(parts[-3:]) return ".".join(parts[-2:]) if len(parts) >= 2 else d def reserved_domain(d): return registrable(d) in RESERVED_DOMAINS or d.rsplit(".", 1)[-1] in RESERVED_TLDS def phone_digits(v): return "" if SCI_RE.match((v or "").strip()) else re.sub(r"\D", "", v or "") def phone_placeholder(v): d = phone_digits(v) if len(d) >= 5 and len(set(d)) == 1: return True return len(d) >= 7 and (d in "01234567890123456789" or d in "98765432109876543210") def phone_key(v, allow_placeholder=False): d = phone_digits(v) if len(d) < 7 or len(d) > 15 or (phone_placeholder(v) and not allow_placeholder): return "" return d[-10:] def sf_suffix(id15): out = "" for i in range(3): mask = 0 for j, c in enumerate(id15[i * 5:(i + 1) * 5]): if "A" <= c <= "Z": mask |= 1 << j out += SF_SUFFIX[mask] return out def sf_shaped(v): return bool(re.fullmatch(r"[A-Za-z0-9]{15}|[A-Za-z0-9]{18}", v)) and any(c.isdigit() for c in v) \ and any(c.isalpha() for c in v) def sf_valid18(v): return len(v) == 18 and sf_shaped(v) and sf_suffix(v[:15]) == v[15:] def sf_id_damage(ids): """Salesforce-shaped IDs that are unsafe to join on: (fifteen, bad_suffix, case_collisions), each sorted. Two 15-character IDs equal once lowercased are different records that case-insensitive tools join together; two 18-character IDs equal once lowercased mean the case was damaged; an 18-character ID whose suffix does not match its letter case was altered. Valid 18-character IDs whose first 15 characters differ only in case are different records, so they are never compared on their first 15 characters.""" shaped = sorted({i for i in ids if i and sf_shaped(i)}) groups = defaultdict(set) for i in shaped: groups[(len(i), i.lower())].add(i) fifteen = [i for i in shaped if len(i) == 15] bad = [i for i in shaped if len(i) == 18 and not sf_valid18(i)] coll = [i for i in shaped if len(groups[(len(i), i.lower())]) > 1] return fifteen, bad, coll def has_word(text, words): """True when a word or phrase starts at a word boundary, so 'soft' never matches inside 'Microsoft'. 'soft' must also end there, because 'software' is not a soft bounce.""" return any(re.search(r"\b" + re.escape(w) + (r"\b" if w == "soft" else ""), text) for w in words) def bounce_class(reason): r = (reason or "").lower().replace("_", " ") # enum values such as UNKNOWN_USER m = re.search(r"\b([245])\.(\d{1,3})\.(\d{1,3})\b", r) if m: code = m.group(0) if m.group(1) == "4": return "temporary" if code in ADDRESS_DEAD_CODES: return "address-dead" if code == "5.2.2": return "mailbox-full" if m.group(2) == "7": return "sender-side" for words, cls in ((MAILBOX_FULL_WORDS, "mailbox-full"), (ADDRESS_DEAD_WORDS, "address-dead"), (SENDER_WORDS, "sender-side"), (TEMP_WORDS, "temporary")): if has_word(r, words): return cls return "unknown" def build_countries(): out = {} for entry in COUNTRY_TABLE.replace("\n", "").split(";"): parts = [p.strip() for p in entry.split("|") if p.strip()] for p in parts: out[fold(p).replace(" ", "")] = parts[0] return out def build_states(): out = {} for entry in US_STATES.replace("\n", "").split(";"): code, name = entry.strip()[:2], entry.strip()[3:] out[fold(code)] = code out[fold(name).replace(" ", "")] = code return out COUNTRIES = build_countries() STATES = build_states() def country_code(v): return COUNTRIES.get(fold(v).replace(" ", ""), "") if v else "" def mask(value, kind): v = (value or "").strip() if not v: return "" if kind == "email": local, domain = email_parts(v.lower()) if not local: return v[0] + "***" dparts = domain.split(".") return f"{local[0]}***@{dparts[0][:1]}***." + ".".join(dparts[1:]) if kind == "phone": d = phone_digits(v) return "phone ending " + d[-2:] if d else "***" return " ".join(t[0] + "***" for t in v.split()) def mask_owner(value, unmask=False): """Owner labels can hold an email address; mask it unless --unmask.""" if unmask or not value: return value or "" return re.sub(r"[^\s()<>,;]+@[^\s()<>,;]+", lambda m: mask(m.group(0), "email"), value) def shape(v): s = re.sub(r"[A-Z]+", "A", re.sub(r"[a-z]+", "a", re.sub(r"\d", "9", v))) return s[:24] # ---------- value parsing ---------- TRUE = {"true", "yes", "y", "1", "t", "checked", "active"} FALSE = {"false", "no", "n", "0", "f", "unchecked", "inactive", "deactivated", "disabled"} ISO_RE = re.compile(r"^(\d{4})-(\d{1,2})-(\d{1,2})(?:[T ].*)?$") SLASH_RE = re.compile(r"^(\d{1,2})/(\d{1,2})/(\d{4})(?:[ T].*)?$") def as_bool(s): s = (s or "").strip().lower() return True if s in TRUE else False if s in FALSE else None def as_amount(s): """Float, None when blank, or "ambiguous" when the separators cannot be read safely.""" v = re.sub(r"[^\d,.\-]", "", s or "") if not re.search(r"\d", v): return None if "," in v and "." in v: v = v.replace(".", "").replace(",", ".") if v.rfind(",") > v.rfind(".") else v.replace(",", "") elif "," in v: if re.fullmatch(r"-?\d{1,3}(,\d{3})+", v): v = v.replace(",", "") elif re.fullmatch(r"-?\d+,\d{1,2}", v): v = v.replace(",", ".") else: return "ambiguous" try: return float(v) except ValueError: return "ambiguous" def split_ids(s): return [x for x in re.split(r"[;,|\s]+", s or "") if x] # ---------- populations ---------- def archived(obj, d): """A converted Salesforce lead is an archived source record: only the X checks, F07 and L10 read it.""" return obj == "leads" and bool(d.get("converted")) def base_label(obj): return "leads excluding test and converted records" if obj == "leads" else f"{obj} excluding test records" # ---------- reading and mapping ---------- class InputError(Exception): pass # ---------- results ---------- def idsort(v): return (0, int(v), "") if v.isdigit() else (1, 0, v) def pct(n, d): return f"{100.0 * n / d:.1f}%" if d else "n/a" def rate(n, d): return f"{n} of {d} ({pct(n, d)})" def plural(n, word): return f"{n} {word}" + ("" if n == 1 else "es" if word.endswith(("s", "x")) else "s") # ---------- F: fields ---------- PLACEHOLDER_FIELDS = ["first_name", "last_name", "full_name", "phone", "mobile", "title", "company_name", "name", "domain", "website", "country", "city", "state", "postal_code", "industry", "lead_status", "deal_type", "lost_reason", "next_step", "__ukey"] def is_placeholder(val, field, allowed): v = val.strip().lower() if not v or (allowed and vfold(v) in allowed): return False if field == "country" and country_code(val): return False if field in ("phone", "mobile") and phone_placeholder(val): return True return v in PLACEHOLDER_VALUES or vfold(v) in {"na", "none", "null", "unknown", "tbd", "asdf"} or \ (len(v) >= 2 and len(set(v)) == 1)