"""Context (populations, owners, associations) and the X, F, E, R, O, L and A checks. Works on tables already in memory: no file access, no network calls. """ import re from collections import Counter, defaultdict from hygiene_core import ( DOT0_RE, FIVE_DIGIT_POSTAL, FORMULA_RE, FREEMAIL, ISO_RE, MOJIBAKE_RE, NBSP, OBJECTS, PERSONS, PLACEHOLDER_FIELDS, PLACEHOLDER_LOCALS, REMOVED_RE, ROLE_LOCALS, SCI_RE, SLASH_RE, STATES, TEST_NAMES, TYPO_DOMAINS, ZERO_WIDTH, archived, as_amount, as_bool, base_label, bounce_class, company_norm, country_code, email_parts, email_valid, fold, idsort, is_placeholder, norm_domain, norm_email, pct, person_name, phone_key, rate, registrable, reserved_domain, sf_id_damage, sf_shaped, sf_valid18, split_ids, vfold) from hygiene_io import date_parser, header_index NAMES = { "X01": "duplicate-record-ids", "X02": "id-damage", "X03": "salesforce-id-shape", "X04": "encoding", "X05": "structure", "X06": "blank-column", "X07": "association-truncation", "X08": "coverage", "X09": "mapping", "X10": "formula-leading-values", "F01": "required-missing", "F02": "placeholder-values", "F03": "picklist-drift", "F04": "format-variants", "F05": "spreadsheet-damage", "F06": "whitespace-and-case", "F07": "test-and-internal", "E01": "invalid-email", "E02": "placeholder-email", "E03": "typo-domain", "E04": "role-address", "E05": "personal-email-on-b2b", "E06": "bounced-still-marketable", "E07": "opted-out-but-billed", "E08": "bounce-classes", "R01": "dormant", "R02": "never-engaged", "R03": "address-decay-rate", "D01": "same-unique-key", "D02": "same-email", "D03": "email-variant", "D04": "person-fuzzy", "D05": "company", "D06": "deal", "D07": "lead-matches-contact", "D08": "shared-value", "D09": "rule-breadth-and-chains", "O01": "missing-owner", "O02": "inactive-or-removed-owner", "O03": "unknown-owner", "O04": "integration-owner", "O05": "queue-aging", "L01": "customer-without-evidence", "L02": "won-but-not-customer", "L03": "won-without-amount", "L04": "closed-with-future-date", "L05": "open-past-close", "L06": "open-without-activity", "L07": "stage-not-in-pipeline", "L08": "lifecycle-blank", "L09": "follow-up-conflict", "L10": "conversion-opt-out-drift", "L11": "lost-without-reason", "A01": "contact-without-company", "A02": "open-deal-without-company", "A03": "open-deal-without-contact", "A04": "wrong-company-domain", "A05": "personal-mail-company", "A06": "over-association", "A07": "dangling-reference", } # ---------- context: populations, owners, associations ---------- class Ctx: def __init__(self, args, rules, tables): self.args, self.rules, self.t = args, rules, tables self.crm = args.crm self.th = rules["thresholds"] self.as_of = args.as_of_date self.internal = {d.lower() for d in rules["internal_domains"]} self.testdom = {d.lower() for d in rules["test_domains"]} self.freemail = FREEMAIL | {d.lower() for d in rules["freemail_domains_add"]} self.dot_domains = {d.lower() for d in rules["dot_insensitive_domains"]} self.typos = dict(TYPO_DOMAINS, **{k.lower(): v for k, v in rules["typo_domains"].items()}) self.won = {vfold(s) for s in rules["closed_won_stages"]} self.lost = {vfold(s) for s in rules["closed_lost_stages"]} self.cust_values = {vfold(s) for s in rules["customer_lifecycle_values"]} self.order = [vfold(s) for s in rules["lifecycle_order"]] self.info = {} self.ids = {} self.users, self.queues = None, {} self._load_users() for obj in OBJECTS: if obj in self.t: self.ids[obj] = {} for i, r in enumerate(self.t[obj].rows): self.ids[obj].setdefault(self.key(self.t[obj].rid(r)), i) self._build_info() self._build_assoc() def key(self, v): v = (v or "").strip() if self.crm == "salesforce" and len(v) == 18 and sf_valid18(v): return v[:15] return v def tables(self): return [(o, self.t[o]) for o in OBJECTS if o in self.t] def _load_users(self): u = self.t.get("users") if u is not None: self.users = {} for r in u.rows: rec = {"id": u.v(r, "user_id"), "name": u.v(r, "name"), "email": u.v(r, "email").lower(), "active": as_bool(u.v(r, "active")), "kind": (u.v(r, "kind") or "person").lower()} if rec["kind"] == "queue": self.queues[self.key(rec["id"])] = rec self.queues[rec["name"].lower()] = rec continue for k in (self.key(rec["id"]), rec["name"].lower(), rec["email"]): if k: self.users[k] = rec q = self.t.get("queues") if q is not None: for r in q.rows: rec = {"id": q.v(r, "queue_id"), "name": q.v(r, "name")} self.queues[self.key(rec["id"])] = rec if rec["name"]: self.queues[rec["name"].lower()] = rec def owner_state(self, value): """Set of flags: missing, removed, inactive, unknown, integration, queue, active.""" v = (value or "").strip() if not v: return {"missing"} if REMOVED_RE.match(v): return {"removed"} flags = set() toks = set(re.findall(r"[a-z0-9]+", v.lower())) if toks & {p.lower() for p in self.rules["integration_owner_patterns"]}: flags.add("integration") if self.key(v) in self.queues or v.lower() in self.queues: return flags | {"queue"} if self.users is None: return flags | {"unverified"} u = self.users.get(self.key(v)) or self.users.get(v.lower()) if not u: return flags | {"unknown"} if u["kind"] == "integration": flags.add("integration") flags.add("inactive" if u["active"] is False else "active") return flags def is_test_email(self, domain): reg = registrable(domain) if domain in self.internal or reg in self.internal or domain in self.testdom or reg in self.testdom: return True return reserved_domain(domain) and not self.args.allow_reserved_domains def _build_info(self): for obj, t in self.tables(): rows = [] for r in t.rows: d = {"rid": t.rid(r), "row": r, "owner": t.v(r, "owner"), "source": t.v(r, "source") or t.v(r, "created_by")} d["created"] = (date_parser(t, "created_at", self.args.date_order)[0] or (lambda x: None))(t.v(r, "created_at")) d["ukey"] = t.v(r, "__ukey") if "__ukey" in t.col else "" if obj in PERSONS: self._person(t, r, d) elif obj == "companies": self._company(t, r, d) else: self._deal(t, r, d) rows.append(d) self.info[obj] = rows def _person(self, t, r, d): raw = re.split(r"[;,\s]+", t.v(r, "email")) if t.v(r, "email") else [] extra = re.split(r"[;,\s]+", t.v(r, "additional_emails")) if t.v(r, "additional_emails") else [] d["email_raw"] = t.v(r, "email") d["email"] = norm_email(raw[0]) if raw else "" d["emails"] = [e for e in dict.fromkeys(norm_email(x) for x in raw + extra) if e] d["valid"] = email_valid(d["email"]) if d["email"] else None d["local"], d["domain"] = email_parts(d["email"]) d["reg"] = registrable(d["domain"]) if d["domain"] else "" d["role"] = d["local"] in ROLE_LOCALS d["personal"] = bool(d["reg"]) and (d["domain"] in self.freemail or d["reg"] in self.freemail) first, last = t.v(r, "first_name"), t.v(r, "last_name") d["name"] = person_name(first, last, t.v(r, "full_name")) d["last_name"] = person_name("", last, "") if last else (d["name"].split()[-1] if d["name"] else "") tokens = set(fold(f"{first} {last} {t.v(r, 'full_name')}").split()) d["test"] = bool(tokens & TEST_NAMES) or (bool(d["domain"]) and self.is_test_email(d["domain"])) d["phones"] = [t.v(r, "phone"), t.v(r, "mobile")] d["pkeys"] = sorted({k for k in (phone_key(p) for p in d["phones"]) if k}) d["lifecycle"] = vfold(t.v(r, "lifecycle_stage")) d["customer"] = d["lifecycle"] in self.cust_values d["lead_status"] = vfold(t.v(r, "lead_status")) d["opted_out"] = as_bool(t.v(r, "opted_out")) d["dnc"] = as_bool(t.v(r, "do_not_call")) d["bounce_reason"] = t.v(r, "bounce_reason") d["bounced"] = bool(d["bounce_reason"]) or bool(t.v(r, "bounced_at")) or as_bool(t.v(r, "hard_bounced")) is True d["company_name"] = company_norm(t.v(r, "company_name")) d["converted"] = as_bool(t.v(r, "is_converted")) is True if t.usable("marketing_status"): d["marketing"] = vfold(t.v(r, "marketing_status")) in {"marketing", "marketingcontact", "true", "yes", "1"} else: d["marketing"] = bool(d["email"]) and d["opted_out"] is not True d["marketing"] = d["marketing"] and not d["test"] and not archived(t.obj, d) def _company(self, t, r, d): dom = norm_domain(t.v(r, "domain")) or norm_domain(t.v(r, "website")) d["domain"] = registrable(dom) if dom else "" d["domains"] = {x for x in [d["domain"]] + [registrable(norm_domain(x)) for x in split_ids(t.v(r, "additional_domains"))] if x} d["personal"] = bool(d["domain"]) and d["domain"] in self.freemail d["name"] = company_norm(t.v(r, "name")) d["country"] = country_code(t.v(r, "country")) d["pkey"] = phone_key(t.v(r, "phone")) d["lifecycle"] = vfold(t.v(r, "lifecycle_stage")) d["customer"] = d["lifecycle"] in self.cust_values d["test"] = bool(set(d["name"].split()) & TEST_NAMES) or (bool(d["domain"]) and self.is_test_email(d["domain"])) def _deal(self, t, r, d): d["name"] = fold(t.v(r, "name")) d["amount"] = as_amount(t.v(r, "amount")) pc = date_parser(t, "close_date", self.args.date_order)[0] pa = date_parser(t, "last_activity_at", self.args.date_order)[0] d["close"] = pc(t.v(r, "close_date")) if pc else None d["activity"] = pa(t.v(r, "last_activity_at")) if pa else None d["stage"] = vfold(t.v(r, "stage")) d["pipeline"] = vfold(t.v(r, "pipeline")) won_b, closed_b = as_bool(t.v(r, "is_won")), as_bool(t.v(r, "is_closed")) lost_b = as_bool(t.v(r, "is_lost")) if closed_b is None and (won_b or lost_b): closed_b = True elif closed_b is None and won_b is False and lost_b is False: closed_b = False d["won"] = won_b if won_b is not None else d["stage"] in self.won d["closed"] = closed_b if closed_b is not None else (d["stage"] in self.won or d["stage"] in self.lost) d["lost"] = d["closed"] and not d["won"] d["open"] = not d["closed"] d["deal_type"] = vfold(t.v(r, "deal_type")) d["lost_reason"] = t.v(r, "lost_reason") d["test"] = bool(set(d["name"].split()) & {"test", "dummy"}) def _build_assoc(self): self.assoc = defaultdict(set) # (from_obj, to_obj, from_key) -> set of to_keys self.raw_refs = [] # (from_obj, from_rid, to_obj, to_key) links = [("contacts", "associated_company_ids", "companies"), ("contacts", "associated_deal_ids", "deals"), ("leads", "associated_company_ids", "companies"), ("companies", "associated_contact_ids", "contacts"), ("companies", "associated_deal_ids", "deals"), ("deals", "associated_company_ids", "companies"), ("deals", "associated_contact_ids", "contacts")] for src, field, dst in links: t = self.t.get(src) if t is None or field not in t.col: continue for r, d in zip(t.rows, self.info[src]): if archived(src, d): continue a = self.key(t.rid(r)) for b in split_ids(t.v(r, field)): b = self.key(b) self.assoc[(src, dst, a)].add(b) self.assoc[(dst, src, b)].add(a) self.raw_refs.append((src, t.rid(r), dst, b, field)) roles = self.t.get("contact_roles") if roles is not None: for r in roles.rows: o, c = self.key(roles.v(r, "opportunity_id")), self.key(roles.v(r, "contact_id")) if o and c: self.assoc[("deals", "contacts", o)].add(c) self.assoc[("contacts", "deals", c)].add(o) self.raw_refs.append(("contact_roles", "", "deals", o, "opportunity_id")) self.raw_refs.append(("contact_roles", "", "contacts", c, "contact_id")) def linked(self, src, dst, rid): return self.assoc.get((src, dst, self.key(rid)), set()) def rec(self, obj, k): i = self.ids.get(obj, {}).get(self.key(k)) return None if i is None else self.info[obj][i] def has_assoc_data(self, src, dst): fields = {("contacts", "companies"): [("contacts", "associated_company_ids"), ("companies", "associated_contact_ids")], ("contacts", "deals"): [("contacts", "associated_deal_ids"), ("deals", "associated_contact_ids")], ("companies", "deals"): [("companies", "associated_deal_ids"), ("deals", "associated_company_ids")]} pair = fields.get((src, dst)) or fields.get((dst, src), []) ok = any(o in self.t and self.t[o].usable(f) for o, f in pair) if (src, dst) in (("contacts", "deals"), ("deals", "contacts")) and "contact_roles" in self.t: ok = True return ok class Results: def __init__(self, exemptions): self.checks, self.findings, self.exempt = [], [], {k: set(v) for k, v in exemptions.items()} def add(self, cid, obj, status, num=None, den=None, pop="", thr="", detail="", ids=(), field="", per=None): if status == "PASS" and den == 0: # nothing was checked, so it is not a pass status, detail = "NOT CHECKED", f"{detail} The population is empty, so nothing was checked.".strip() ids = sorted({i for i in ids if i}, key=idsort) gone = [i for i in ids if i in self.exempt.get(cid, set())] if gone: ids = [i for i in ids if i not in gone] if num is not None and num >= len(gone): num -= len(gone) detail += f" Exempt per rules file: {len(gone)} record(s)." if not ids and status in ("FAIL", "WARN"): status = "PASS" self.checks.append({"id": cid, "name": NAMES[cid], "object": obj, "field": field, "status": status, "numerator": num, "denominator": den, "population": pop, "threshold": thr, "detail": detail.strip(), "record_ids": ids}) for i in ids: self.findings.append((cid, status, obj, i, field, (per or {}).get(i, NAMES[cid]))) def nc(self, cid, obj, reason, field=""): self.add(cid, obj, "NOT CHECKED", detail=reason, field=field) def na(self, cid, obj, reason, field=""): self.add(cid, obj, "N/A", detail=reason, field=field) # ---------- X: export integrity and coverage ---------- def checks_x(ctx, res): idfields = ["record_id", "associated_company_ids", "associated_deal_ids", "associated_contact_ids", "converted_contact_id", "parent_id", "phone", "mobile", "__ukey"] for obj, t in ctx.tables(): ids = [t.rid(r) for r in t.rows] cnt = Counter(i for i in ids if i) dups = [i for i, c in cnt.items() if c > 1] blank = sum(1 for i in ids if not i) res.add("X01", obj, "FAIL" if dups or blank else "PASS", len(dups) + blank, len(t.rows), "rows", "any", f"{len(dups)} record IDs repeated, {blank} rows without an ID.", dups) bad = {} for f in idfields: for r in t.rows: for val in (split_ids(t.v(r, f)) if f.startswith("associated") else [t.v(r, f)]): if val and (SCI_RE.match(val) or DOT0_RE.match(val)): bad[t.rid(r) or f"row {t.rows.index(r) + 2}"] = f"{f} looks converted to a number" res.add("X02", obj, "FAIL" if bad else "PASS", len(bad), len(t.rows), "rows", "any", "Scientific notation or a trailing .0 in ID-like columns." if bad else "", list(bad), per=bad) shaped = [i for i in ids if i and sf_shaped(i)] if ctx.crm == "salesforce" or (ids and len(shaped) >= 0.9 * len([i for i in ids if i])): fifteen, badsuf, coll = sf_id_damage(shaped) st = "FAIL" if badsuf or coll else "WARN" if fifteen else "PASS" res.add("X03", obj, st, len(set(fifteen) | set(badsuf) | set(coll)), len(shaped), "Salesforce-shaped IDs", "15-character IDs WARN; FAIL for IDs of the same length equal once lowercased, or an 18-character " "suffix that does not match its letter case", f"{len(fifteen)} 15-character, {len(coll)} case collisions, {len(badsuf)} inconsistent suffixes.", set(fifteen) | set(badsuf) | set(coll)) else: res.na("X03", obj, "IDs are not Salesforce IDs.") moji = [t.rid(r) for r in t.rows if any(MOJIBAKE_RE.search(c) for c in r)] st = "WARN" if moji or t.encoding != "utf-8" else "PASS" res.add("X04", obj, st, len(moji), len(t.rows), "rows", "any", f"Decoded as {t.encoding}; {len(moji)} rows with mojibake or replacement characters.", moji) res.add("X05", obj, "FAIL" if t.malformed else "PASS", len(t.malformed), len(t.rows), "rows", "any", ("Rows with the wrong number of fields: " + ", ".join(map(str, t.malformed[:10]))) if t.malformed else "") blank = sorted({f"{t.headers[i]} ({f})" for f, i in t.col.items() if i in t.blank_cols and not f.startswith("__")} | {f"{t.headers[i]} (rule column)" for i in t.extra.values() if i in t.blank_cols}) if "__ukey" in t.col and t.col["__ukey"] in t.blank_cols: blank.append(f"{t.headers[t.col['__ukey']]} (unique key)") res.add("X06", obj, "WARN" if blank else "PASS", len(blank), None, "mapped column(s) blank on every row", "any", ("NOT RETRIEVED: " + "; ".join(blank) + ". Dependent checks are NOT CHECKED.") if blank else "") trunc = [t.rid(r) for r in t.rows for f in ("associated_company_ids", "associated_deal_ids", "associated_contact_ids") if len(split_ids(t.v(r, f))) >= 1000] res.add("X07", obj, "WARN" if trunc else "PASS", len(trunc), len(t.rows), "rows", "1,000 IDs in one cell", "Association cells at the default export cap; export All associated records." if trunc else "", trunc) exp = ctx.args.expected.get(obj) if exp is None: res.add("X08", obj, "NOT CHECKED", len(t.rows), None, "export rows", "equal to the CRM count", "No --expected-count given; coverage UNCONFIRMED.") else: st = "PASS" if exp == len(t.rows) else "FAIL" res.add("X08", obj, st, len(t.rows), exp, "CRM record count", "equal", "Coverage FULL." if st == "PASS" else "Coverage PARTIAL: explain the difference or re-export.") amb = [f for f, v in t.mapinfo.items() if v[0] == "ambiguous"] res.add("X09", obj, "WARN" if amb else "PASS", len(amb), None, "ambiguous mapping(s)", "none", ("Ambiguous: " + "; ".join(f"{f} ({t.mapinfo[f][2]})" for f in amb)) if amb else f"{len([f for f in t.col if not f.startswith('__')])} fields mapped; see the mapping report.") formula = [t.rid(r) for r in t.rows if any(FORMULA_RE.match(c) for c in r)] res.add("X10", obj, "WARN" if formula else "PASS", len(formula), len(t.rows), "rows", "any", "Cells that a spreadsheet would run as formulas." if formula else "", formula) def live(ctx, obj, test_too=False): """The records every population starts from: no converted leads, and no test records unless test_too.""" return [d for d in ctx.info[obj] if not archived(obj, d) and (test_too or not d.get("test"))] def population(ctx, obj, name): base = live(ctx, obj) if name in ("all", ""): return base, None if name == "marketing" and obj in PERSONS: return [d for d in base if d["marketing"]], None if name == "customers": return [d for d in base if d.get("customer")], None if obj != "deals" else "deals have no lifecycle" if name == "owner_required" and obj in PERSONS: req = {vfold(s) for s in ctx.rules["owner_required_lifecycle"]} return [d for d in base if d.get("lifecycle") in req], None if obj == "deals" and name in ("open", "won", "lost", "closed"): return [d for d in base if d[name]], None if name == "unconverted" and obj in PERSONS: return [d for d in base if not d["converted"]], None return [], f"unknown population '{name}'" def checks_f(ctx, res): th, allowed_all = ctx.th, ctx.rules["allowed_values"] for obj, t in ctx.tables(): crit = {e["field"] for e in ctx.rules["required"].get(obj, []) if e.get("critical")} for e in ctx.rules["required"].get(obj, []): f, popname = e["field"], e.get("population", "all") idx = header_index(t, f) if idx is None: res.nc("F01", obj, f"column '{f}' is not in the export", field=f) continue if idx in t.blank_cols: res.nc("F01", obj, f"'{t.headers[idx]}' was not retrieved (blank on every row); see X06", field=f) continue recs, err = population(ctx, obj, popname) if err: res.nc("F01", obj, err, field=f) continue allowed = {vfold(x) for x in allowed_all.get(f"{obj}.{f}", [])} miss = [d["rid"] for d in recs if not d["row"][idx].strip() or is_placeholder(d["row"][idx], f, allowed)] n, den = len(miss), len(recs) if e.get("critical"): st = "FAIL" if n else "PASS" elif den and n / den > th["non_critical_missing_fail_share"] and n >= th["non_critical_missing_fail_min_records"]: st = "FAIL" else: st = "WARN" if n else "PASS" thr = "any (critical)" if e.get("critical") else (f"WARN at 1; FAIL above {th['non_critical_missing_fail_share']:.0%}" f" and {th['non_critical_missing_fail_min_records']} records") popl = f"{popname} {obj}" + (" excluding converted" if obj == "leads" else "") res.add("F01", obj, st, n, den, popl, thr, f"{t.headers[idx]} missing: {rate(n, den)}.", miss, field=f) if not ctx.rules["required"].get(obj): res.nc("F01", obj, "no required fields defined for this object in the rules file") base = live(ctx, obj) per = {} for d in base: for f in PLACEHOLDER_FIELDS: if f in t.col: allowed = {vfold(x) for x in allowed_all.get(f"{obj}.{f}", [])} if is_placeholder(d["row"][t.col[f]], f, allowed): per[d["rid"]] = (per.get(d["rid"], "placeholder in") + f" {f}") res.add("F02", obj, "WARN" if per else "PASS", len(per), len(base), base_label(obj), "any", "Placeholder values counted apart from blanks." if per else "", per, per=per) entries = {k.split(".", 1)[1]: v for k, v in allowed_all.items() if k.split(".", 1)[0] == obj} if not entries: res.nc("F03", obj, "no allowed values defined for this object in the rules file") for f, vals in entries.items(): idx = header_index(t, f) if idx is None or idx in t.blank_cols: res.nc("F03", obj, f"column '{f}' is not in the export or is blank", field=f) continue allowed = {vfold(x) for x in vals} filled = [d for d in base if d["row"][idx].strip()] off = {d["rid"]: f"value outside the allowed list in {f}" for d in filled if vfold(d["row"][idx]) not in allowed} st = ("FAIL" if f in crit else "WARN") if off else "PASS" res.add("F03", obj, st, len(off), len(filled), f"{obj} with a value in {f}", "any", "", off, field=f, per=off) if obj != "deals": checks_f04_f05(ctx, res, obj, t, base) per6, warn6 = {}, False for d in base: for f, i in t.col.items(): c = d["row"][i] if not c: continue if NBSP in c or c != c.translate(ZERO_WIDTH): per6[d["rid"]] = f"invisible character in {f}" warn6 = warn6 or f in ("email", "domain", "website", "record_id", "__ukey", "additional_emails") elif c != c.strip() or " " in c: per6.setdefault(d["rid"], f"extra spaces in {f}") for f in ("first_name", "last_name", "full_name"): if any(len(w) >= 4 and w.isalpha() and w.isupper() for w in t.v(d["row"], f).split()): per6.setdefault(d["rid"], f"all-caps {f}") res.add("F06", obj, "WARN" if warn6 else "INFO" if per6 else "PASS", len(per6), len(base), base_label(obj), "WARN for invisible characters in email, domain or ID fields", "", per6, per=per6) if obj in PERSONS or obj == "companies": tests = [d["rid"] for d in ctx.info[obj] if d.get("test")] res.add("F07", obj, "WARN" if tests else "PASS", len(tests), len(t.rows), f"all {obj} rows", "any", "Test and internal records, excluded from every other denominator." if tests else "", tests) def checks_f04_f05(ctx, res, obj, t, base): if not (t.usable("country") or t.usable("state")): res.nc("F04", obj, t.why("country") or "no country or state column") else: per, notes, den = {}, [], 0 for f, table in (("country", None), ("state", STATES)): if not t.usable(f): continue spell = defaultdict(Counter) for d in base: raw = t.v(d["row"], f) if not raw: continue den += 1 code = country_code(raw) if f == "country" else table.get(fold(raw).replace(" ", ""), "") if code: spell[code][raw] += 1 for code, c in sorted(spell.items()): if len(c) > 1: canon = sorted(c.items(), key=lambda kv: (-kv[1], kv[0]))[0][0] notes.append(f"{f} {code}: " + ", ".join(f"'{k}' {n}" for k, n in sorted(c.items())) + f"; propose '{canon}'") for d in base: raw = t.v(d["row"], f) if raw in c and raw != canon: per[d["rid"]] = f"{f} variant '{raw}', propose '{canon}'" res.add("F04", obj, "WARN" if per else "PASS", len(per), den, f"{obj} values in country or state", "any", "; ".join(notes), per, per=per) per = {} for d in base: pc, cc = t.v(d["row"], "postal_code"), country_code(t.v(d["row"], "country")) if cc in FIVE_DIGIT_POSTAL and re.fullmatch(r"\d{4}", pc): per[d["rid"]] = f"postal code has 4 digits for {cc} (leading zero lost)" for f in ("phone", "mobile"): if SCI_RE.match(t.v(d["row"], f)): per[d["rid"]] = f"{f} in scientific notation" mixed = [] for f in ("created_at", "close_date", "last_engagement_at", "last_activity_at"): if t.usable(f): vals = [t.v(r, f) for r in t.rows if t.v(r, f)] if any(ISO_RE.match(v) for v in vals) and any(SLASH_RE.match(v) for v in vals): mixed.append(f) res.add("F05", obj, "WARN" if per or mixed else "PASS", len(per), len(base), base_label(obj), "any", ("Mixed date formats in " + ", ".join(mixed) + ". ") if mixed else "", per, per=per) # ---------- E: email and reachability, R: decay ---------- def checks_e_r(ctx, res): th = ctx.th for obj in PERSONS: if obj not in ctx.t: continue t, allrows = ctx.t[obj], live(ctx, obj, test_too=True) base = [d for d in allrows if not d["test"]] mkt = [d for d in base if d["marketing"]] mpop = "marketing contacts" if t.usable("marketing_status") else "records with an email, not opted out" if not t.usable("email"): for cid in ("E01", "E02", "E03", "E04", "E05", "E06", "E07", "E08", "R01", "R02", "R03"): res.nc(cid, obj, t.why("email")) continue withmail = [d for d in base if d["email_raw"]] bad = [d for d in withmail if not d["valid"]] st = "FAIL" if any(d["marketing"] for d in bad) else "WARN" if bad else "PASS" res.add("E01", obj, st, len(bad), len(withmail), f"{obj} with an email", "FAIL at 1 in the marketing population", f"{sum(1 for d in bad if d['marketing'])} in the marketing population.", [d["rid"] for d in bad]) every = [d for d in allrows if d["email"]] ph = {d["rid"]: "placeholder address" for d in every if d["local"] in PLACEHOLDER_LOCALS or d["domain"] in ctx.testdom or (d["domain"] and reserved_domain(d["domain"]) and not ctx.args.allow_reserved_domains)} res.add("E02", obj, "WARN" if ph else "PASS", len(ph), len(every), f"every {obj} row with an email, test rows included" + (", converted excluded" if obj == "leads" else ""), "any", "", ph, per=ph) ty = {d["rid"]: f"domain looks like a typo of {ctx.typos[d['domain']]} (draft correction only)" for d in withmail if d["domain"] in ctx.typos} res.add("E03", obj, "WARN" if ty else "PASS", len(ty), len(withmail), f"{obj} with an email", "any", "", ty, per=ty) roles = [d["rid"] for d in withmail if d["role"]] res.add("E04", obj, "INFO" if roles else "PASS", len(roles), len(withmail), f"{obj} with an email", "", "Role mailboxes are excluded from person matching." if roles else "", roles) if ctx.rules["b2b"]: pers = [d["rid"] for d in withmail if d["personal"]] res.add("E05", obj, "INFO" if pers else "PASS", len(pers), len(withmail), f"{obj} with an email", "", "Personal domains never corroborate a company match." if pers else "", pers) else: res.na("E05", obj, "rules.b2b is false") if not (t.usable("bounce_reason") or t.usable("hard_bounced") or t.usable("bounced_at")): res.nc("E06", obj, t.why("bounce_reason") + " and no bounced flag") else: b = [d["rid"] for d in mkt if d["bounced"]] note = ("An empty bounce reason never proves an address is sendable (HubSpot global bounces)." if ctx.crm == "hubspot" else "Bounce evidence read: a bounce reason or a bounced flag.") res.add("E06", obj, "FAIL" if b else "PASS", len(b), len(mkt), mpop, "any", note, b) if not t.usable("marketing_status"): res.na("E07", obj, "no marketing contact status column (HubSpot billing concept)") elif not t.usable("opted_out"): res.nc("E07", obj, t.why("opted_out")) else: o = [d["rid"] for d in mkt if d["opted_out"]] res.add("E07", obj, "WARN" if o else "PASS", len(o), len(mkt), mpop, "any", "Opted out of all email yet billed as marketing contacts." if o else "", o) classes = Counter(bounce_class(d["bounce_reason"]) for d in base if d["bounce_reason"]) if not t.usable("bounce_reason"): res.nc("E08", obj, t.why("bounce_reason")) else: per = {d["rid"]: bounce_class(d["bounce_reason"]) for d in base if d["bounce_reason"]} hand = " Sender-side bounces go to deliverability (email-sender-infrastructure)." if classes["sender-side"] else "" res.add("E08", obj, "INFO" if per else "PASS", len(per), len(base), base_label(obj), "", ", ".join(f"{k} {v}" for k, v in sorted(classes.items())) + "." + hand if per else "", per, per=per) eng = [f for f in ("last_engagement_at", "last_form_submission_at", "last_click_at") if t.usable(f)] parsers = [date_parser(t, f, ctx.args.date_order)[0] for f in eng] if not eng or not any(parsers): res.nc("R01", obj, "no engagement date column (last engagement, last form submission, last click)") res.nc("R02", obj, "no engagement date column") else: def latest(d): ds = [p(t.v(d["row"], f)) for f, p in zip(eng, parsers) if p] ds = [x for x in ds if x] return max(ds) if ds else None dorm = [d["rid"] for d in mkt if latest(d) and (ctx.as_of - latest(d)).days > th["dormant_days"]] share = len(dorm) / len(mkt) if mkt else 0 res.add("R01", obj, "WARN" if share >= th["dormant_warn_share"] else "PASS", len(dorm), len(mkt), mpop, f"WARN at {th['dormant_warn_share']:.0%}", f"No engagement in {th['dormant_days']} days: {rate(len(dorm), len(mkt))}. " "Report for a sunset decision; never delete from this audit.", dorm) old = [d for d in mkt if d["created"] and (ctx.as_of - d["created"]).days > th["never_engaged_days"]] never = [d["rid"] for d in old if not latest(d)] if not t.usable("created_at"): res.nc("R02", obj, t.why("created_at")) else: res.add("R02", obj, "WARN" if never else "PASS", len(never), len(old), f"{mpop} created more than {th['never_engaged_days']} days ago", "any", "", never) if not t.usable("bounce_reason"): res.nc("R03", obj, t.why("bounce_reason")) else: dead = sum(1 for d in base if d["bounce_reason"] and bounce_class(d["bounce_reason"]) == "address-dead") if t.usable("send_count"): den = sum(1 for d in base if (as_amount(t.v(d["row"], "send_count")) or 0) not in ("ambiguous", 0)) popn = f"{obj} with at least one send" else: den, popn = len(withmail), f"{obj} with an email" res.add("R03", obj, "INFO", dead, den, popn, "trend metric", f"Address-dead bounces: {rate(dead, den)}.") # ---------- O: ownership ---------- def owned_through_company(ctx, d): """contact_owner_from_company: an associated company owned by an active person (or one no users file can verify).""" for k in ctx.linked("contacts", "companies", d["rid"]): c = ctx.rec("companies", k) if not c or c["test"] or not c["owner"]: continue f = ctx.owner_state(c["owner"]) if f & {"active", "unverified"} and not f & {"integration", "queue", "removed", "inactive", "unknown"}: return True return False def checks_o(ctx, res): th = ctx.th req = {vfold(s) for s in ctx.rules["owner_required_lifecycle"]} crit_objs = set(ctx.rules["owner_critical_customer_objects"]) open_co = set() if "deals" in ctx.t: for d in ctx.info["deals"]: if d["open"] and not d["test"]: open_co |= ctx.linked("deals", "companies", d["rid"]) for obj, t in ctx.tables(): base = live(ctx, obj) if not t.usable("owner"): for cid in ("O01", "O02", "O03", "O04"): res.nc(cid, obj, t.why("owner")) if obj == "leads": res.nc("O05", obj, t.why("owner")) continue def critical(d): return (obj == "deals" and d["open"]) or (obj in crit_objs and d.get("customer", False)) if obj in PERSONS: warnpop = [d for d in base if (d["lifecycle"] in req) or (obj == "leads" and not d["converted"])] elif obj == "companies": warnpop = [d for d in base if ctx.key(d["rid"]) in open_co] else: warnpop = [] failpop = [d for d in base if critical(d)] pop = {d["rid"]: d for d in failpop + warnpop} miss = [d for d in pop.values() if not d["owner"]] note = "" if obj == "contacts" and ctx.rules["contact_owner_from_company"]: if "companies" in ctx.t and ctx.has_assoc_data("contacts", "companies"): via = {d["rid"] for d in miss if owned_through_company(ctx, d)} miss = [d for d in miss if d["rid"] not in via] note = f" Owned through an associated company (contact_owner_from_company): {len(via)}." else: note = " contact_owner_from_company not applied: needs the companies export with contact to company associations." fails = [d for d in miss if critical(d)] gap = (" Companies with open deals not determined: no deal to company association columns." if obj == "companies" and "deals" in ctx.t and not ctx.has_assoc_data("companies", "deals") else "") demoted = obj in ("contacts", "companies") and obj not in crit_objs res.add("O01", obj, "FAIL" if fails else "WARN" if miss else "PASS", len(miss), len(pop), f"owner-required {obj}", "WARN at 1; customers here are not in owner_critical_customer_objects" if demoted else "FAIL at 1 for open deals and customers; WARN at 1 otherwise", f"Open deals or customers: {len(fails)}. Other owner-required records: {len(miss) - len(fails)}.{gap}{note}", [d["rid"] for d in miss]) owned = [d for d in base if d["owner"]] flags = {d["rid"]: ctx.owner_state(d["owner"]) for d in owned} gone = [d for d in owned if flags[d["rid"]] & {"removed", "inactive"}] if ctx.users is None and not gone: res.nc("O02", obj, "needs a users file with active status") else: f2 = [d for d in gone if critical(d)] res.add("O02", obj, "FAIL" if f2 else "WARN" if gone else "PASS", len(gone), len(owned), f"{obj} with an owner", "WARN; customers here are not in owner_critical_customer_objects" if demoted else "FAIL for open deals and customers; WARN otherwise", f"Open deals or customers: {len(f2)}." + (" Partial: no users file, only removed-user labels read." if ctx.users is None else ""), [d["rid"] for d in gone]) if ctx.users is None: res.nc("O03", obj, "needs a users file") else: unk = [d["rid"] for d in owned if "unknown" in flags[d["rid"]]] res.add("O03", obj, "WARN" if unk else "PASS", len(unk), len(owned), f"{obj} with an owner", "any", "Not in the users or queues list: check queues and stale IDs." if unk else "", unk) integ = [d for d in owned if "integration" in flags[d["rid"]]] st = "WARN" if integ else "PASS" detail = "" if obj == "deals": od = [d for d in base if d["open"]] io = [d for d in integ if d["open"]] if od and len(io) / len(od) >= th["integration_owner_fail_share"]: st = "FAIL" detail = f"Open deals owned by an integration: {rate(len(io), len(od))}." res.add("O04", obj, st, len(integ), len(owned), f"{obj} with an owner", f"WARN at 1; FAIL at {th['integration_owner_fail_share']:.0%} of open deals" if obj == "deals" else "any", detail, [d["rid"] for d in integ]) if obj == "leads": if ctx.crm != "salesforce": res.na("O05", obj, "queue aging applies to Salesforce leads") elif not ctx.queues: res.nc("O05", obj, "needs queues (users file rows with kind queue, or --queues)") else: aged = [d["rid"] for d in base if not d["converted"] and "queue" in flags.get(d["rid"], set()) and d["created"] and (ctx.as_of - d["created"]).days > th["queue_max_days"]] res.add("O05", obj, "WARN" if aged else "PASS", len(aged), len(base), "unconverted leads", f"queue-owned for more than {th['queue_max_days']} days", "Hand off to lead-routing-and-speed-to-lead." if aged else "", aged) # ---------- L: lifecycle and stages ---------- def checks_l(ctx, res): th, rules = ctx.th, ctx.rules deals = ctx.t.get("deals") dinfo = [d for d in ctx.info.get("deals", []) if not d["test"]] won_ids = {ctx.key(d["rid"]) for d in dinfo if d["won"]} co_won = set() for d in dinfo: if d["won"]: co_won |= ctx.linked("deals", "companies", d["rid"]) evid_field = rules["customer_evidence_field"] for obj in ("contacts", "companies"): if obj not in ctx.t: continue t, base = ctx.t[obj], [d for d in ctx.info[obj] if not d["test"]] if not t.usable("lifecycle_stage"): res.nc("L01", obj, t.why("lifecycle_stage")) if obj == "companies": res.nc("L02", obj, t.why("lifecycle_stage")) continue if deals is None: res.nc("L01", obj, "needs the deals export") if obj == "companies": res.nc("L02", obj, "needs the deals export") continue if obj == "companies": linked_ok, gap = ctx.has_assoc_data("companies", "deals"), "no company to deal association columns" else: linked_ok = ctx.has_assoc_data("contacts", "deals") or ( ctx.has_assoc_data("contacts", "companies") and ctx.has_assoc_data("companies", "deals")) gap = "no contact to deal association columns, and no contact to company plus company to deal columns" if not linked_ok: res.nc("L01", obj, gap + "; export with associations") if obj == "companies": res.nc("L02", obj, gap + "; export with associations") continue ei = header_index(t, evid_field) if evid_field else None cust = [d for d in base if d["customer"]] def evidence(d): if ei is not None and as_bool(d["row"][ei]): return True if obj == "companies": return bool(ctx.linked(obj, "deals", d["rid"]) & won_ids) if ctx.linked(obj, "deals", d["rid"]) & won_ids: return True return bool(ctx.linked(obj, "companies", d["rid"]) & co_won) no = [d["rid"] for d in cust if not evidence(d)] share = len(no) / len(cust) if cust else 0 res.add("L01", obj, "FAIL" if share > th["customer_without_evidence_fail_share"] else "WARN" if no else "PASS", len(no), len(cust), f"customer-stage {obj}", f"WARN at 1; FAIL above {th['customer_without_evidence_fail_share']:.0%}", f"No closed-won deal (direct or through the company) and no customer flag: {rate(len(no), len(cust))}.", no) if obj == "companies": ci = ctx.order.index(next(iter(ctx.cust_values))) if ctx.cust_values and next(iter(ctx.cust_values)) in ctx.order else None early = [d["rid"] for d in base if ctx.key(d["rid"]) in co_won and not d["customer"] and (not d["lifecycle"] or (ci is not None and d["lifecycle"] in ctx.order and ctx.order.index(d["lifecycle"]) < ci))] res.add("L02", obj, "WARN" if early else "PASS", len(early), len(co_won & set(ctx.ids["companies"])), "companies with a closed-won deal", "any", "Check the lifecycle automation for won deals." if early else "", early) if deals is not None: t = deals won = [d for d in dinfo if d["won"]] if not t.usable("amount"): res.nc("L03", "deals", t.why("amount")) else: zero_ok = {vfold(x) for x in rules["zero_amount_deal_types"]} noamt = [d["rid"] for d in won if (d["amount"] is None or d["amount"] == 0) and d["deal_type"] not in zero_ok] amb = [d["rid"] for d in won if d["amount"] == "ambiguous"] res.add("L03", "deals", "FAIL" if noamt else "PASS", len(noamt), len(won) - len(amb), "closed-won deals", "any", (f"{len(amb)} amounts with unreadable separators were not judged; re-export plain numbers." if amb else ""), noamt) pclose, prob = date_parser(t, "close_date", ctx.args.date_order) if not pclose: for cid in ("L04", "L05"): res.nc(cid, "deals", prob) else: closed = [d for d in dinfo if d["closed"] and d["close"]] fut = [d["rid"] for d in closed if (d["close"] - ctx.as_of).days > 1] res.add("L04", "deals", "FAIL" if fut else "PASS", len(fut), len(closed), "closed deals with a close date", "any", "", fut) opn = [d for d in dinfo if d["open"]] past = [d for d in opn if d["close"] and d["close"] < ctx.as_of] stale = [d for d in past if (ctx.as_of - d["close"]).days > th["stale_open_deal_days_past_close"]] st = "FAIL" if opn and len(stale) / len(opn) > th["stale_open_deal_fail_share"] else "WARN" if past else "PASS" res.add("L05", "deals", st, len(past), len(opn), "open deals", f"FAIL when more than {th['stale_open_deal_fail_share']:.0%} are over {th['stale_open_deal_days_past_close']} days past", f"{len(stale)} more than {th['stale_open_deal_days_past_close']} days past close ({pct(len(stale), len(opn))}).", [d["rid"] for d in past]) if not t.usable("last_activity_at"): res.nc("L06", "deals", t.why("last_activity_at")) else: opn = [d for d in dinfo if d["open"]] lim = th["no_activity_days"] idle = [d["rid"] for d in opn if (d["activity"] and (ctx.as_of - d["activity"]).days > lim) or (not d["activity"] and d["created"] and (ctx.as_of - d["created"]).days > lim)] res.add("L06", "deals", "WARN" if idle else "PASS", len(idle), len(opn), "open deals", f"no activity in {lim} days", "", idle) if not rules["pipelines"]: res.nc("L07", "deals", "no pipelines and stages in the rules file") elif not t.usable("stage"): res.nc("L07", "deals", t.why("stage")) else: pipes = {vfold(k): {vfold(s) for s in v} for k, v in rules["pipelines"].items()} single = next(iter(pipes.values())) if len(pipes) == 1 else None checked, orphan, unlisted = 0, [], 0 for d in dinfo: stages = pipes.get(d["pipeline"]) if t.usable("pipeline") else single if stages is None: unlisted += 1 continue checked += 1 if d["stage"] not in stages: orphan.append(d["rid"]) note = "" if t.usable("pipeline") else "No pipeline column: checked against the only pipeline in the rules file. " res.add("L07", "deals", "FAIL" if orphan else "PASS", len(orphan), checked, "deals in listed pipelines", "any", note + (f"{unlisted} deals in pipelines not listed in the rules file." if unlisted else ""), orphan) if not t.usable("lost_reason"): res.nc("L11", "deals", t.why("lost_reason")) elif not rules["lost_reason_required"]: res.na("L11", "deals", "rules.lost_reason_required is false") else: lost = [d for d in dinfo if d["lost"]] nr = [d["rid"] for d in lost if not d["lost_reason"]] res.add("L11", "deals", "WARN" if nr else "PASS", len(nr), len(lost), "closed-lost deals", "any", "", nr) if "contacts" in ctx.t: t, base = ctx.t["contacts"], [d for d in ctx.info["contacts"] if not d["test"]] if not t.usable("lifecycle_stage"): res.nc("L08", "contacts", t.why("lifecycle_stage")) res.nc("L09", "contacts", t.why("lifecycle_stage")) else: old = [d for d in base if d["created"] and (ctx.as_of - d["created"]).days > 1] blank = [d["rid"] for d in old if not d["lifecycle"]] res.add("L08", "contacts", "WARN" if blank else "PASS", len(blank), len(old), "contacts created before yesterday", "any", "", blank) if not t.usable("lead_status"): res.nc("L09", "contacts", t.why("lead_status")) else: openv = {vfold(s) for s in rules["open_lead_status_values"]} cust = [d for d in base if d["customer"]] conf = [d["rid"] for d in cust if d["lead_status"] in openv] res.add("L09", "contacts", "WARN" if conf else "PASS", len(conf), len(cust), "customer-stage contacts", "any", "", conf) if "leads" in ctx.t: if ctx.crm != "salesforce": res.na("L10", "leads", "conversion opt-out drift applies to Salesforce") elif "contacts" not in ctx.t: res.nc("L10", "leads", "needs the contacts export") elif not (ctx.t["leads"].usable("is_converted") and ctx.t["leads"].usable("converted_contact_id")): res.nc("L10", "leads", ctx.t["leads"].why("is_converted", "converted_contact_id")) else: conv = [d for d in ctx.info["leads"] if d["converted"] and not d["test"]] drift = {} for d in conv: c = ctx.rec("contacts", ctx.t["leads"].v(d["row"], "converted_contact_id")) if not c: continue if d["opted_out"] and c["opted_out"] is not True: drift[d["rid"]] = f"Email Opt Out lost on contact {c['rid']}" elif d["dnc"] and c["dnc"] is not True: drift[d["rid"]] = f"Do Not Call lost on contact {c['rid']}" res.add("L10", "leads", "FAIL" if drift else "PASS", len(drift), len(conv), "converted leads", "any", "", drift, per=drift) # ---------- A: associations ---------- def checks_a(ctx, res): if "contacts" in ctx.t: base = [d for d in ctx.info["contacts"] if not d["test"]] if not ctx.has_assoc_data("contacts", "companies"): for cid in ("A01", "A04", "A06"): res.nc(cid, "contacts", "no contact to company association columns; export with associations") else: elig = [d for d in base if ctx.rules["b2b"] and d["valid"] and not d["personal"] and not d["role"]] no = [d["rid"] for d in elig if not ctx.linked("contacts", "companies", d["rid"])] res.add("A01", "contacts", "WARN" if no else "PASS", len(no), len(elig), "B2B contacts with a valid work email", "any", "", no) if "companies" not in ctx.t: res.nc("A04", "contacts", "needs the companies export") else: owners = defaultdict(set) for c in ctx.info["companies"]: for dom in c["domains"]: if dom not in ctx.freemail: owners[dom].add(ctx.key(c["rid"])) wrong = {} for d in elig: linked = ctx.linked("contacts", "companies", d["rid"]) if linked and d["reg"] in owners and not (linked & owners[d["reg"]]): wrong[d["rid"]] = f"email domain belongs to company {', '.join(sorted(owners[d['reg']], key=idsort))}" res.add("A04", "contacts", "WARN" if wrong else "PASS", len(wrong), len(elig), "B2B contacts with a valid work email", "any", "Wrong association or a job change." if wrong else "", wrong, per=wrong) many = [d["rid"] for d in base if len(ctx.linked("contacts", "companies", d["rid"])) > 5] res.add("A06", "contacts", "WARN" if many else "PASS", len(many), len(base), "contacts", "more than 5 companies", "", many) if "deals" in ctx.t: dinfo = [d for d in ctx.info["deals"] if not d["test"]] opn = [d for d in dinfo if d["open"]] if not ctx.has_assoc_data("deals", "companies"): res.nc("A02", "deals", "no deal to company association columns") else: noc = [d["rid"] for d in opn if not ctx.linked("deals", "companies", d["rid"])] req = ctx.rules["deal_company_required"] res.add("A02", "deals", ("FAIL" if req else "WARN") if noc else "PASS", len(noc), len(opn), "open deals", "any", "", noc) if ctx.crm == "salesforce" and "contact_roles" not in ctx.t: res.nc("A03", "deals", "needs an OpportunityContactRole export (--contact-roles)") elif not ctx.has_assoc_data("deals", "contacts"): res.nc("A03", "deals", "no deal to contact association columns") else: nop = [d["rid"] for d in opn if not ctx.linked("deals", "contacts", d["rid"])] res.add("A03", "deals", "WARN" if nop else "PASS", len(nop), len(opn), "open deals", "any", "", nop) if not ctx.has_assoc_data("deals", "companies"): res.nc("A06", "deals", "no deal to company association columns") else: many = [d["rid"] for d in dinfo if len(ctx.linked("deals", "companies", d["rid"])) > 3] res.add("A06", "deals", "WARN" if many else "PASS", len(many), len(dinfo), "deals", "more than 3 companies", "", many) if "companies" in ctx.t: base = [d for d in ctx.info["companies"] if not d["test"]] pm = [d["rid"] for d in base if d["personal"]] res.add("A05", "companies", "WARN" if pm else "PASS", len(pm), len(base), "companies", "any", "Usually an import, integration, enrichment tool or manual entry wrote a personal domain; " "find the writer with the record source." if pm else "", pm) dangling = defaultdict(dict) for src, src_rid, dst, key, field in ctx.raw_refs: if dst in ctx.t and key not in ctx.ids.get(dst, {}) and src_rid: dangling[src][src_rid] = f"{field} points to {dst} {key}, absent from the export" for obj, t in ctx.tables(): refs = [r for r in ctx.raw_refs if r[0] == obj] if not refs: res.nc("A07", obj, "no association IDs in this export") continue missing = sorted({r[2] for r in refs if r[2] not in ctx.t}) per = dangling.get(obj, {}) res.add("A07", obj, "WARN" if per else "PASS", len(per), len({r[1] for r in refs}), f"{obj} with association IDs", "any", ("Not checked against: " + ", ".join(missing) + " (file not given). ") if missing else "", per, per=per)