"""Shared parts of the crm-audit rule engine: value helpers, writer labels, Res (one rule's result) and Ctx (the merged inventory; Ctx.writers() gathers every writer of one field). audit_crm.py loads this file and the three rule modules by path and registers each under its file name, so the rule modules can import from it; it is not run on its own. Standard library only, no network, Python 3.9 or later. """ import datetime as dt import re STATUS_ORDER = {"FAIL": 3, "WARN": 2, "PASS": 1} # Allowed values per enum key (references/inventory-format.md). Values are compared ignoring case, # spaces, hyphens and underscores and rewritten to the form below; "a | b" left from a template stops the run. ENUM_VALUES = { "crm": ["hubspot", "salesforce", "attio", "pipedrive", "zoho", "dynamics", "other"], "business_model": ["b2b", "b2c", "self-serve"], "scope": ["full", "gates"], "coverage_status": ["complete", "partial", "unavailable"], "stage_type": ["open", "won", "lost"], "account_type": ["employee", "contractor", "agency", "shared", "generic", "integration", "former_employee"], "integration_type": ["data_sync", "enrichment", "event_app", "no_code", "reverse_etl", "middleware", "private_app", "other"], "identity_type": ["personal", "dedicated", "private_app", "service_key", "oauth_app", "integration_user"], "direction": ["into_crm", "out_of_crm", "two_way"], "blank_semantics": ["preserve", "clear", "undefined"], "stage_mapping": ["ids", "labels"], "writer_kind": ["form-setting", "setting", "workflow", "integration", "billing", "enrichment", "import"], "manual_status": ["PASS", "WARN", "FAIL", "NOT CHECKED", "N/A"], "evidence_kind": ["writer sample", "record trace", "reconciliation", "access view", "settings capture"], } HUMAN_TYPES = {"user", "human", "manual", "crm-ui", "ui", "workspace-member", "form"} CRITICAL_WRITER_CLASSES = {"routing", "ownership", "lifecycle", "stage", "revenue", "identity"} ACTION_NAMES = {"0-18224765": "Delete contact", "0-31": "Set marketing contact status", "0-43347357": "Manage communication subscriptions", "0-177946906": "Enrich record", "0-172351286": "Data Agent: Custom prompt", "0-216647524": "Data Agent: Research", "0-201091202": "Data Agent: Fill Smart Property", "0-219292676": "Enroll in prospecting agent", "delete_records": "Delete Records element", # Named in reports only: normalize_exports.py lists these as unclassified, because each sends a # message, writes to another system or writes a property the v4 JSON does not name. "0-18": "Set Salesforce campaign", "0-19": "Create Salesforce task", "0-25085031": "Send WhatsApp message", "0-40900952": "Send SMS message", "0-199186210": "Send survey", "0-219160146": "Track intent signals", "0-219161394": "Stop tracking intent signals", "0-225935194": "Validate and format phone number", "0-195318603": "Summarize record", "0-207702619": "Infer company value proposition and ICP", "0-218870680": "Use a custom LLM"} # The inventory keys that show an automation was read for each flag (an empty list means "read, none"). KNOWN_KEYS = {"writes_known": ("writes",), "actions_known": ("destructive_actions", "consent_actions", "action_types"), "refs_known": ("filter_refs",)} def action_label(t): """An action type ID with its name when the scripts know one: '0-40900952 Send SMS message'.""" return "%s %s" % (t, ACTION_NAMES[t]) if t in ACTION_NAMES else str(t) def num(v): if v is None or v == "" or isinstance(v, bool): return None try: return float(v) except (TypeError, ValueError): return None def slug(s): return re.sub(r"[^a-z0-9]+", "-", str(s or "").strip().lower()).strip("-") def alnum(s): return re.sub(r"[^a-z0-9]", "", str(s or "").lower()) def norm_writer(label): label = str(label or "").strip() if not label: return "unknown" t, n = (label.split(":", 1) + [""])[:2] if ":" in label else (label, "") t, n = slug(t), slug(n) if t in HUMAN_TYPES and t != "form": return "user" + (":" + n if n else "") return t + (":" + n if n else "") def is_automated(label): return label.split(":", 1)[0] not in HUMAN_TYPES def pct(n, d): return "%.1f%%" % (100.0 * n / d) def fmt_num(v): return "{:,.0f}".format(v) if float(v).is_integer() else "{:,.2f}".format(v) def ratio(c): if not isinstance(c, dict): return None n, d = num(c.get("numerator")), num(c.get("denominator")) return None if n is None or d is None else (n, d) def parse_date(s): s = str(s or "").strip() if not s: return None m = re.match(r"^(\d{4})-(\d{2})-(\d{2})", s) if m: try: return dt.date(int(m.group(1)), int(m.group(2)), int(m.group(3))) except ValueError: return None try: v = float(s) return dt.datetime.fromtimestamp(v / (1000.0 if v > 1e11 else 1.0), tz=dt.timezone.utc).date() except ValueError: return None def tokens(name): s = re.sub(r"([a-z])([A-Z])", r"\1 \2", str(name or "")) return [t for t in re.split(r"[^a-z0-9]+", s.lower()) if t] def empty(v): return v is None or (isinstance(v, (str, list, dict)) and len(v) == 0) or (isinstance(v, str) and not v.strip()) def worst(items): best = None for st, _ in items: if st in STATUS_ORDER and (best is None or STATUS_ORDER[st] > STATUS_ORDER[best]): best = st return best def prob(s): p = num(s.get("probability")) return None if p is None else (p / 100.0 if p > 1 else p) def norm_name(c, name): n = str(name or "").lower() for s in c.cfg.get("field_suffixes_to_strip") or []: if n.endswith(s): n = n[: -len(s)] break for p in c.cfg.get("field_prefixes_to_strip") or []: if n.startswith(p) and len(n) > len(p): n = n[len(p):] break words = set(c.cfg.get("field_words_to_strip") or []) toks = [t for t in re.split(r"[^a-z]+", re.sub(r"\d+", " ", n)) if t and t not in words] return "".join(toks), set(toks) class Res(object): """One rule's result: findings (status, text), NOT CHECKED reasons, notes, N/A reason and class used. partial lists the parts of the rule that were not checked. A gate whose status is PASS or WARN but which has a NOT CHECKED reason or a partial entry cannot reach a ready verdict (audit_crm.verdicts). gap() records a part that also stops the rule passing: once the rule has run, close() turns a PASS into NOT CHECKED and keeps the passing findings as notes; FAIL and WARN findings stand. part() records a part that leaves the status as it is (an AUTO-06 PASS "among classified actions").""" def __init__(self): self.items, self.nc_reasons, self.na, self.cls, self.notes = [], [], None, None, [] self.partial, self.gaps = [], [] def add(self, status, text): self.items.append((status, text)) def nc(self, reason): self.nc_reasons.append(reason) def gap(self, reason, kept="rated before this gap: "): self.gaps.append((reason, kept)) def part(self, note): self.notes.append(note) self.partial.append(note) def close(self): for reason, _ in self.gaps: self.nc_reasons.append(reason) self.partial.append(reason) if self.gaps and worst(self.items) == "PASS": self.notes.extend(self.gaps[0][1] + t for _, t in self.items) self.items = [] self.gaps = [] def status(self): if self.na: return "N/A" return worst(self.items) or "NOT CHECKED" class Ctx(object): def __init__(self, inv, cfg, as_of): self.inv, self.cfg, self.as_of = inv, cfg, as_of self.meta = inv.get("meta") or {} self.crm = str(self.meta.get("crm") or "").lower() self.coverage = inv.get("coverage") or {} self.settings = inv.get("settings") or {} self.counts = inv.get("counts") or {} self.decisions = inv.get("decisions") or {} self.reporting = inv.get("reporting") self.lifecycle = inv.get("lifecycle") self.fields = inv.get("fields") self.automations = inv.get("automations") self.integrations = inv.get("integrations") self.users = inv.get("users") self.pipelines = inv.get("pipelines") self.objects = inv.get("objects") self.history = {(str(h.get("object") or "").lower(), str(h.get("field") or "").lower()): h for h in (inv.get("history") or []) if isinstance(h, dict)} self.owner_fields = set(x.lower() for x in cfg.get("owner_fields") or []) self.field_index, self.field_by_name = {}, {} for f in self.fields or []: k = (str(f.get("object") or "").lower(), str(f.get("name") or "").lower()) self.field_index[k] = f self.field_by_name.setdefault(k[1], []).append(f) self.critical = self._critical() self.crit_index = {(c["object"], c["field"].lower()): c for c in self.critical} self.crit_by_name = {} for c in self.critical: self.crit_by_name.setdefault(c["field"].lower(), []).append(c) self.lifecycle_field = str((self.lifecycle or {}).get("field") or {"hubspot": "lifecyclestage"}.get(self.crm, "lifecyclestage")).lower() # Decision IDs that a decision_ref or a manual downgrade may cite: decisions.items entries with a date. self.dated_decisions = set(str(d.get("id")) for d in self.decisions.get("items") or [] if isinstance(d, dict) and not empty(d.get("id")) and parse_date(d.get("date"))) # History exports often name a workflow or app by ID (workflow:585051946). Map those to the label the # inventory uses for the same automation or integration, so one writer is not counted twice. self.alias = {} for a in self.automations or []: if not empty(a.get("id")): for prefix in ("workflow", "flow", "process", "automation"): self.alias["%s:%s" % (prefix, slug(a["id"]))] = self.automation_label(a) for i in self.integrations or []: for key in ("id", "app_id"): if not empty(i.get(key)): for prefix in ("integration", "app", "api"): self.alias["%s:%s" % (prefix, slug(i[key]))] = self.integration_label(i) def label(self, writer): lab = norm_writer(writer) return self.alias.get(lab, lab) def unclassified(self): """(automation, [action type IDs]) for enabled automations with actions the scripts cannot classify.""" return [(a.get("name") or a.get("id"), [str(x) for x in a.get("unknown_action_types")]) for a in self.enabled() if a.get("unknown_action_types")] def stage_type(self, s): """HubSpot works out closed status from probability (1.0 closed won, 0.0 closed lost); other CRMs record it separately (Salesforce IsClosed and IsWon), so the recorded type is used.""" p = prob(s) if self.crm == "hubspot" and p is not None: return "won" if p >= 1.0 else ("lost" if p <= 0.0 else "open") return str(s.get("type") or "open").lower() def _critical(self): merged = {} for src in (self.cfg.get("critical_fields") or [], self.inv.get("critical_fields") or []): for c in src: if not isinstance(c, dict) or not c.get("field") or str(c["field"]).startswith("<"): continue k = (str(c.get("object") or "").lower(), str(c["field"]).lower()) cur = merged.setdefault(k, {"object": k[0], "field": str(c["field"])}) for key, v in c.items(): if key in ("object", "field"): continue if key == "class" and cur.get("class"): continue if not empty(v) or key not in cur: cur[key] = v return list(merged.values()) def days_since(self, s): d = parse_date(s) return None if d is None else (self.as_of - d).days def enabled(self): """Automations that are on, plus hand-entered ones whose state was not recorded: they may run.""" return [a for a in self.automations or [] if a.get("enabled") in (True, "true", "True", 1, None, "")] def is_known(self, a, flag): """True when the flag is true, or when it is unset and the entry carries a list the flag covers (KNOWN_KEYS; an empty list means "read, none"). A flag that is false, or unset with none of those lists, means the automation's writes, actions or references were never read.""" v = a.get(flag) if v is True or str(v).lower() == "true": return True return v is None and any(isinstance(a.get(k), (list, dict)) for k in KNOWN_KEYS[flag]) def enabled_known(self, flag): """(known, unknown): enabled automations split by whether their writes, actions or references were read.""" act = self.enabled() return [a for a in act if self.is_known(a, flag)], [a for a in act if not self.is_known(a, flag)] def automation_label(self, a): if a.get("writer_label"): return norm_writer(a["writer_label"]) prefix = {"workflow": "workflow", "flow": "flow", "process_builder": "process", "workflow_rule": "workflow-rule", "apex": "apex", "flow_other": "flow"}.get(a.get("layer") or "workflow", slug(a.get("layer")) or "automation") return prefix + ":" + slug(a.get("name") or a.get("id")) def integration_label(self, i): return norm_writer(i.get("writer_label") or "integration:" + str(i.get("name") or i.get("id"))) def int_fields(self, i): """(object or None, field) pairs an integration writes.""" objs = [str(o).lower() for o in (i.get("objects") or [])] out = [] for f in i.get("fields_written") or []: f = str(f) if "." in f: o, n = f.split(".", 1) out.append((o.lower(), n.lower())) else: out.append((objs[0] if len(objs) == 1 else None, f.lower())) return out def writes_int(self, i): return bool(i.get("creates_records") or i.get("fields_written") or i.get("access") == "write" or str(i.get("direction") or "") in ("into_crm", "two_way")) def is_critical(self, obj, field): return self.critical_for(obj, field) is not None def critical_for(self, obj, field): field = str(field or "").lower() if obj and (str(obj).lower(), field) in self.crit_index: return self.crit_index[(str(obj).lower(), field)] hits = self.crit_by_name.get(field) or [] return hits[0] if hits else None def writers(self, obj, field): """Observed writers of one field from every source, label -> set of evidence sources.""" f, o = str(field).lower(), (str(obj).lower() if obj else None) out = {} def add(label, src): out.setdefault(self.label(label), set()).add(src) for a in self.enabled(): ao = str(a.get("object") or "").lower() or None for w in a.get("writes") or []: wo = str(w.get("object") or "").lower() or ao # a Create record action writes the object it creates if str(w.get("field") or "").lower() == f and (w.get("associated") or not (o and wo and wo != o)): add(self.automation_label(a), "automation definition") if o and ao and ao != o: continue for w in a.get("owner_writes") or []: if str(w.get("field") or "").lower() == f: add(self.automation_label(a), "automation definition") for i in self.integrations or []: for io, name in self.int_fields(i): if name == f and (not o or not io or io == o): add(self.integration_label(i), "integration mapping") if f == self.lifecycle_field and isinstance(self.lifecycle, dict): for w in self.lifecycle.get("writers") or []: if isinstance(w, dict) and w.get("writer"): add(w["writer"], "lifecycle settings") for fm in self.inv.get("forms") or []: if fm.get("sets_lifecycle"): add("form-setting:" + slug(fm.get("name")), "form settings") for fd in self.fields or []: # the writers column of a generic field export (Pipedrive, Zoho, Dynamics) fo = str(fd.get("object") or "").lower() or None if str(fd.get("name") or "").lower() == f and (not o or not fo or fo == o): for w in fd.get("writers") or []: add(w, "field export") c = self.critical_for(o, f) for w in (c or {}).get("observed_writers") or []: add(w, "observed") h = self.history.get((o or "", f)) or next((v for k, v in self.history.items() if k[1] == f), None) for w in (h or {}).get("changes_by_writer", {}) or {}: add(w, "history") return out def section_state(self, name): c = self.coverage.get(name) if isinstance(c, dict): return str(c.get("status") or "").lower(), c.get("reason") or c.get("note") or "" if isinstance(c, str): return c.lower(), "" return "", "" def hist_for(c, f): return c.history.get((f["object"], f["field"].lower())) or next( (v for k, v in c.history.items() if k[1] == f["field"].lower()), None) UNKNOWN_HINT = {"writes_known": "field writes", "actions_known": "actions", "refs_known": "filter and branch references"} def names_of(autos, limit=6): n = [repr(str(a.get("name") or a.get("id"))) for a in autos] return ", ".join(n[:limit]) + (" and %d more" % (len(n) - limit) if len(n) > limit else "") def known_or_nc(c, r, flag): """Enabled automations whose writes, actions or references were read. NOT CHECKED when none were; when only some were, the rest are a NOT CHECKED part (Res.gap), so a PASS turns NOT CHECKED.""" known, unknown = c.enabled_known(flag) if not unknown: return known what = ("%s unknown for %d of %d enabled automations: %s; read each full definition (HubSpot workflows get, " "Salesforce flow metadata or Org Check, the CRM's automation screen), record it in the inventory and set " "%s to true" % (UNKNOWN_HINT[flag], len(unknown), len(known) + len(unknown), names_of(unknown), flag)) if not known: r.nc(what) return None r.gap(what, "for the automations read: ") return known def unknown_writes_note(c, r): _, unknown = c.enabled_known("writes_known") if unknown: r.part("field writes unknown for %d enabled automations (%s), so the writer list may be incomplete" % ( len(unknown), names_of(unknown))) def permitted_ok(label, permitted): if label in permitted: return True if label.split(":", 1)[0] == "user" and any(p.split(":", 1)[0] == "user" and ":" not in p for p in permitted): return True return False def check_ratio(r, c, what, warn=None, fail=None, below=False, noun=""): """Rate one count object. above: FAIL when share > fail; below: FAIL when share < fail.""" rt = ratio(c) if rt is None: return None n, d = rt if d == 0: r.nc("%s: zero denominator" % what) return "zero" share = n / d pop = (" (population: %s)" % c.get("population")) if c.get("population") else "" base = "%s of %s %s (%s)%s" % (fmt_num(n), fmt_num(d), noun or what, pct(n, d), pop) if below: if fail is not None and share < fail: r.add("FAIL", "%s; fail below %.0f%%" % (base, 100 * fail)) elif warn is not None and share < warn: r.add("WARN", "%s; warn below %.0f%%" % (base, 100 * warn)) else: r.add("PASS", base) else: if fail is not None and share > fail: r.add("FAIL", "%s; fail above %.0f%%" % (base, 100 * fail)) elif warn is not None and share > warn: r.add("WARN", "%s; warn above %.0f%%" % (base, 100 * warn)) else: r.add("PASS", base) return "rated"