"""Drift between the plan and the observed inventories (references/drift-rules.md). Run through audit_plan.py drift. This module holds the setup, the observed name classes, rules D01 to D06 with the D12 occurrence keys, the report and the --out writer. The Drift class takes the property, identifier and registration rules from drift_properties.py and the reference counts and cross-tool matrix from drift_counts.py. """ import csv import json import math import os import re from collections import OrderedDict, defaultdict from datetime import date, timedelta from common import (CLASSES, DEFAULT_TRANSPORT_NOTE, DEFINITION_SYNONYMS, GENERATED_RE, LATE_DAYS, PRINT_ORDER, SETTLED_DAYS, REDACTION_TOKENS, RULE_NAMES, SENTINEL_NOTE, SEVERITY_ORDER, SKILL_DIR, TEST_TOKENS, TOOL_COLUMNS, TRANSPORT_NOTES, Inventory, Plan, fail, first_items, ga4_event_name_problem, is_vendor, lines_text, load_vendor, mask, masked_example, norm_key, num, parse_date, pct, plural, read_csv, reserved_lists, src_list, summary_counts) from drift_counts import CountRules from drift_properties import PropertyRules from plan_quality import quality_rules, verdict # ---------- drift setup, name resolution and classes ---------- class Drift(PropertyRules, CountRules): def __init__(self, args): self.args = args self.plan = Plan(args.events, args.properties) repeated = ([f"event {n} ({lines_text(lines)})" for n, lines in self.plan.repeated_events().items()] + [f"property {e}.{p} ({lines_text(lines)})" for (e, p), lines in self.plan.repeated_props().items()]) if repeated: fail(f"the plan defines {first_items(repeated, 4)} more than once, so drift results would depend on " "which row wins. Keep one row for each (quality lists them under Q07) and run drift again.") self.inv = Inventory(args.observed) self.vendor_events = load_vendor(args.vendor_events) self.vendor_params = load_vendor(args.vendor_params) self.reserved = {"events": reserved_lists(self.vendor_events), "params": reserved_lists(self.vendor_params)} self.reference = (read_csv(args.reference, ["event", "tool", "count"], "reference counts")[1] if args.reference else None) self.definitions = self.read_definitions(args.ga4_definitions) if args.ga4_definitions else None wanted = [t.strip().lower() for t in (args.tool or "").split(",") if t.strip()] self.tools = wanted or self.inv.tools() self.findings, self.classes, self.matrix = [], OrderedDict(), [] self.ws, self.we = parse_date(args.window_start or ""), parse_date(args.window_end or "") if (args.window_start and not self.ws) or (args.window_end and not self.we): fail("--window-start and --window-end must be YYYY-MM-DD") if self.ws and self.we and self.ws > self.we: fail("--window-start is after --window-end") if args.as_of and not parse_date(args.as_of): fail("--as-of must be YYYY-MM-DD") # Without --as-of the audit date is today, but never later than 4 days after the window end, so a # window that ended yesterday is still judged as filling. self.as_of_given = bool(args.as_of) self.as_of = parse_date(args.as_of or "") or ( min(date.today(), self.we + timedelta(days=SETTLED_DAYS)) if self.we else date.today()) self.days = (self.we - self.ws).days + 1 if (self.ws and self.we) else None self.lag = (self.as_of - self.we).days if self.we else None self.provisional = self.lag is not None and self.lag < SETTLED_DAYS self.lag_text = (f"ends {plural(self.lag, 'day')} before the as-of date" if self.lag and self.lag > 0 else "runs up to or past the as-of date") self.resolved = {} self.ratios = defaultdict(dict) self.checked = defaultdict(int) # how many items each rule evaluated self.unlabelled = defaultdict(list) # D06: events whose rows carry no source label self.stream_platform = defaultdict(list) # D06: GA4 events planned from a server self.no_dates = defaultdict(list) # D01: events with rows but no last_seen dates self.pii_rows = defaultdict(lambda: [0, 0]) # D13 per tool: property rows checked, rows with pattern counts self.tz_missing = [] # D12: reference rows without both time zones self.no_band = 0 # D12: ratios reported without a documented band, so not judged @staticmethod def read_definitions(path): _, rows = read_csv(path, ["parameter_name", "scope"], "GA4 custom definitions", DEFINITION_SYNONYMS) out = defaultdict(list) for row in rows: if row["parameter_name"]: out[row["parameter_name"]].append(((row.get("scope", "") or "event").strip().upper(), (row.get("kind", "") or "dimension").strip().lower())) return out def add(self, rid, severity, tool, event, prop, message, disposition="", source="", affected=None, denominator=None, unit="", last_seen=None): share = (affected / denominator) if (affected is not None and denominator) else None self.findings.append({ "rule": rid, "name": RULE_NAMES.get(rid, ""), "severity": severity, "tool": tool, "source": source, "event": event, "property": prop, "affected": affected, "unit": unit if affected is not None else "", "denominator": denominator, "share_pct": round(share * 100, 2) if share is not None else None, "last_seen": str(last_seen) if last_seen else "", "message": message, "default_disposition": disposition, }) def tier(self, e): return e["tier"] if e["tier"] is not None else 2 def expected_sources(self, e, tool): return e["overrides"].get(tool) or e["sources"] def resolve(self, tool): planned, aliases, mapped = {}, defaultdict(list), {} for e in self.plan.events.values(): fn, has_column = self.plan.final_name(e, tool) mapped[e["name"]] = (fn, has_column) if fn: planned.setdefault(fn, e["name"]) observed = self.inv.names(tool) keyed = {norm_key(fn): fn for fn in planned} for n in observed: if n not in planned and not self.inv.events[tool].get(n) and norm_key(n) in keyed: aliases[keyed[norm_key(n)]].append(n) # warehouse table name for a planned event self.resolved[tool] = (planned, aliases, mapped) return planned, aliases, mapped def obs_names(self, tool, final_name): planned, aliases, _ = self.resolved[tool] return [final_name] + aliases.get(final_name, []) def plan_event_for(self, tool, name): planned, aliases, _ = self.resolved[tool] if name in planned: return self.plan.events.get(planned[name]) for fn, names in aliases.items(): if name in names: return self.plan.events.get(planned[fn]) return None def classify(self, tool): planned, aliases, mapped = self.resolved[tool] alias_names = {a for v in aliases.values() for a in v} canon = {} for e in self.plan.events.values(): canon[e["name"]] = e["name"] for t in TOOL_COLUMNS: fn = self.plan.final_name(e, t)[0] if fn: canon.setdefault(fn, e["name"]) cls, notes = {}, {} rest = [] planned_keys = defaultdict(list) for fn in planned: planned_keys[norm_key(fn)].append(fn) for n in sorted(self.inv.names(tool)): if n in planned or n in alias_names: cls[n] = "planned" elif is_vendor(n, tool, self.vendor_events): cls[n] = "vendor" elif norm_key(n) in planned_keys: cls[n] = "variant" notes[n] = planned_keys[norm_key(n)][0] elif GENERATED_RE.match(n): cls[n] = "generated" # a CDP destination's per-page event name else: rest.append(n) groups = defaultdict(list) for n in rest: if mask(n) != n: groups[mask(n)].append(n) for pattern, members in groups.items(): if len(members) >= self.args.dynamic_min: for n in members: cls[n] = "dynamic" notes[n] = pattern rest = [n for n in rest if n not in cls] keyed = defaultdict(list) for n in rest: keyed[norm_key(n)].append(n) for n in rest: if TEST_TOKENS & set(norm_key(n).split("_")): cls[n] = "test" elif len(keyed[norm_key(n)]) > 1: cls[n] = "variant" notes[n] = [m for m in keyed[norm_key(n)] if m != n][0] else: cls[n] = "orphan" if n in canon: e = self.plan.events[canon[n]] fn = mapped.get(e["name"], (None, False))[0] notes[n] = (f"is planned event {e['name']}, which the plan maps to {tool} as '{fn}'" if fn else f"is planned event {e['name']}, which the plan does not send to {tool}") return cls, notes def ranks(self, tool): totals = sorted(((self.inv.total(tool, [n]), n) for n in self.inv.names(tool)), reverse=True) totals = [(t, n) for t, n in totals if t > 0] return {n: i + 1 for i, (t, n) in enumerate(totals)}, len(totals) def run(self): quality = quality_rules(self.plan, self.args.ga4_param_limit, self.reserved) self.verdict = verdict(quality, self.plan) self.blocking = [r for r in quality if r["id"] in ("Q05", "Q07", "Q08")] for tool in self.tools: if tool not in self.inv.tools(): self.add("D01", "NOT CHECKED", tool, "", "", f"no inventory rows for {tool}; nothing was compared.") audited = [t for t in self.tools if t in self.inv.tools()] for tool, column in TOOL_COLUMNS.items(): if tool in audited or column not in self.plan.event_columns: continue mapped = [e["name"] for e in self.plan.events.values() if self.plan.final_name(e, tool)[0]] if mapped: verb = "maps" if len(mapped) == 1 else "map" self.add("D01", "NOT CHECKED", tool, "", "", f"{plural(len(mapped), 'planned event')} {verb} to {tool} but no inventory was given " f"({first_items(mapped, 4)}).") if audited and not self.days: self.add("D01", "NOT CHECKED", "", "", "", "no --window-start and --window-end, so events that stopped inside the window were not checked.") no_ids = [] for tool in audited: self.resolve(tool) cls, notes = self.classify(tool) self.classes[tool] = (cls, notes) for rid in ("D02", "D03", "D04"): self.checked[rid] += len(cls) if TOOL_COLUMNS.get(tool) not in self.plan.event_columns: self.add("D01", "NOT CHECKED", tool, "", "", f"no {TOOL_COLUMNS.get(tool, tool + '_name')} column; " "planned names were compared as canonical names.") self.check_events(tool, cls, notes) self.check_properties(tool, cls) if not self.check_identity(tool): no_ids.append(tool) if no_ids: self.add("D14", "NOT CHECKED", ", ".join(no_ids), "", "", "no identifier rows; run the profiler with " "--identity-keys or the identifier queries to classify IDs without printing them.") for tool, labels in self.unlabelled.items(): self.add("D06", "NOT CHECKED", tool, "", "", f"no source label on some or all rows of {plural(len(labels), 'event')} ({first_items(labels, 4)}); " "planned sources were not compared. Add the platform dimension (GA4 Data API), keep the source " "column of the queries, or profile with a source field.") for tool, labels in self.stream_platform.items(): self.add("D06", "INFO", tool, "", "", f"GA4 labels every row with the data stream platform (web, ios or android), so the planned " f"server route of {first_items(labels, 4)} cannot be told apart from a browser or app producer " "here; compare with the system of record (D12) or a parameter only the server sends.") for tool, labels in self.no_dates.items(): self.add("D01", "NOT CHECKED", tool, "", "", f"no last_seen dates for {plural(len(labels), 'event')} ({first_items(labels, 4)}); stops inside " "the window were not checked. Add first_seen and last_seen (GA4 Data API: add the date dimension " "and keep the latest date per event and platform).") for tool in audited: checked, with_counts = self.pii_rows[tool] if not with_counts: what = (f"no pattern counts on its {plural(checked, 'property row')}, so only property names were " "checked" if checked else "no property rows, so property names and values were not checked") self.add("D13", "NOT CHECKED", tool, "", "", f"{what}; run the profiler, GA4 Part B, Segment Query 2b or PostHog Query 1 for pii_hits.") self.check_registration(audited) self.check_reference(audited) self.check_matrix(audited) for rid in RULE_NAMES: if any(f["rule"] == rid and f["severity"] in ("FAIL", "WARN") for f in self.findings): continue if self.checked[rid]: self.add(rid, "PASS", "", "", "", f"{plural(self.checked[rid], 'check')}; " + self.pass_text(rid, audited)) elif not any(f["rule"] == rid for f in self.findings): self.add(rid, "NOT CHECKED", "", "", "", "nothing in the plan or the inventory for this rule to evaluate.") self.findings.sort(key=lambda f: (SEVERITY_ORDER[f["severity"]], PRINT_ORDER.index(f["rule"]), f["tool"], f["event"], f["property"])) for i, f in enumerate([f for f in self.findings if f["severity"] in ("FAIL", "WARN", "INFO")], start=1): f["id"] = f"F{i:02d}" def pass_text(self, rid, audited): """What a PASS line says was checked; parts that did not run are left out and print NOT CHECKED apart.""" limit = self.args.ga4_param_limit stops = self.days and not any(self.no_dates.values()) ratios = [] if self.checked["ratios"]: judged = self.checked["ratios"] - self.no_band if judged and self.no_band: ratios.append(f"{plural(judged, 'event count')} {'matches' if judged == 1 else 'match'} the reference") elif judged: ratios.append("event counts match the reference counts") if self.no_band: one = self.no_band == 1 ratios.append(f"{plural(self.no_band, 'ratio')} {'has' if one else 'have'} no documented band and " f"{'was' if one else 'were'} reported, not judged") if self.checked["dup_keys"]: ratios.append("no occurrence key repeats") patterns = [t for t in audited if self.pii_rows[t][1]] if patterns and len(patterns) == len([t for t in audited if self.pii_rows[t][0]]): pii = "no property named like personal data and no email, phone or sensitive URL patterns" elif patterns: pii = (f"no property named like personal data, and no email, phone or sensitive URL patterns in " f"{', '.join(patterns)}") else: pii = "names only: no property named like personal data" texts = { "D01": "every implemented or verified event mapped to an audited tool has rows in the window" + (" and kept sending to its end" if stops else ""), "D02": "no unplanned events outside vendor, variant, generated and dynamic names", "D03": "no spelling variants of planned or observed names", "D04": f"no group of {self.args.dynamic_min} or more names or keys that differ only by IDs or numbers, " "and no per-page event names", "D05": "no deprecated or retired event has rows, and no approved event is already live", "D06": "every labelled source is planned for its tool, and every planned source sends", "D07": f"required properties are present on 100% of rows ({SENTINEL_NOTE})", "D08": f"every optional property appears on events with {num(self.args.optional_min_rows)} or more rows", "D09": "no unplanned properties on planned events", "D10": "observed value types match the plan types", "D11": "observed enum values are all allowed", "D12": ("; " if self.no_band else " and ").join(ratios) or "nothing failed", "D13": pii, "D14": " and ".join(x for x in (f"GA4 names are 40 characters or fewer and events plan {limit} parameters " "or fewer" if "ga4" in audited else "", "identifier classes and null shares are clean" if self.checked["id_rows"] else "") if x) or "nothing failed", "D15": "every planned GA4 parameter that needs a custom definition is registered with the planned scope, " "and built-in parameters and identifiers need none", } return texts[rid] + f" ({', '.join(audited) or 'no tools'})." # ---------- events, D01 to D06 and D12 keys ---------- def check_events(self, tool, cls, notes): inv = self.inv planned, aliases, mapped = self.resolved[tool] for e in self.plan.events.values(): fn, has_column = mapped[e["name"]] if not fn: continue names = self.obs_names(tool, fn) label = e["name"] if fn == e["name"] else f"{fn} (plan {e['name']})" if not any(inv.has_counts(tool, n) for n in names) and any(inv.listed(tool, n) for n in names): what = (f"listed by {tool} without counts (names only)" if any(inv.events[tool].get(n) for n in names) else "seen as a table or column list only") self.add("D01", "INFO", tool, e["name"], "", f"{label}: {what}, so D01, D05 and D06 were not checked.") continue total = inv.total(tool, names) last = inv.last_seen(tool, names) status, tier = e["status"], self.tier(e) self.checked["D01"] += 1 self.checked["D05"] += 1 if status in ("deprecated", "retired"): if total > 0: sev = "FAIL" if status == "retired" else "WARN" self.add("D05", sev, tool, e["name"], "", f"{label}: status {status} but {num(total)} rows in the window from " f"{src_list(inv.by_source(tool, names))}. Check whether the rows come from old app " "versions (version tail rule) before fixing.", "fix implementation, or a status correction recorded as a decision", affected=total, unit="rows in the window", last_seen=last) continue if status in ("proposed", "approved"): if total > 0: self.add("D05", "WARN", tool, e["name"], "", f"{label}: status {status} but {num(total)} rows are already collected; the plan lags " "the release.", "amend plan (status to implemented, verified after its acceptance tests pass)", affected=total, unit="rows in the window", last_seen=last) else: self.add("D01", "INFO", tool, e["name"], "", f"{label}: status {status}, not shipped yet; no rows " "expected.") continue if total == 0: self.report_unseen(tool, e, fn, has_column, label, last, cls, notes) continue self.check_stopped(tool, e, label, names, total) self.check_sources(tool, e, label, names, total) self.check_duplicates(tool, e, label, names, total) self.check_unplanned(tool, cls, notes, planned) def report_unseen(self, tool, e, fn, has_column, label, last, cls, notes): inv, status = self.inv, e["status"] extra = "" if tool == "ga4" and len(fn) > 40: extra += (f" Root cause: the planned GA4 name is {len(fn)} characters and GA4 does not log event names " "over 40.") problem = ga4_event_name_problem(fn, self.reserved["events"]) if tool == "ga4" else None if problem: extra += (f" Root cause: the planned GA4 name {problem}, and GA4's event naming rules say such names " "can't be used; confirm in DebugView.") variants = [n for n, c in cls.items() if c == "variant" and notes.get(n) == fn and inv.total(tool, [n]) > 0] if variants: extra += f" A spelling variant '{variants[0]}' carries {num(inv.total(tool, [variants[0]]))} rows (D03)." if not has_column: extra += " Compared by canonical name; the tool's name column is missing." if last: self.add("D01", "FAIL", tool, e["name"], "", f"{label}: stopped. Last seen {last}, no rows from {self.ws or 'the window start'} to " f"{self.we or 'the window end'}; status {status}.{extra}", "fix implementation; verified becomes implemented until the acceptance tests pass again" if status == "verified" else "fix implementation", affected=0, unit="rows in the window", last_seen=last) return age = "" if e["introduced_at"]: days = (self.as_of - e["introduced_at"]).days age = (f" In the plan for {days} days" + ("; older than 90 days, so the owner decides its status" if days > 90 else "") + ".") sev = "FAIL" if status == "verified" or self.tier(e) == 1 else "WARN" self.add("D01", sev, tool, e["name"], "", f"{label}: never seen in the inventory; status {status or 'unknown'}, tier {e['tier'] or 'unknown'}." f"{age}{extra} Search every emitter root before calling it not emitted.", "needs decision (fix the emitter, or move the status back to approved)", affected=0, unit="rows in the window") def check_stopped(self, tool, e, label, names, total): """D01: rows in the window, then a silence before its end that the event's volume does not explain.""" if not self.days: return pad = max(0, SETTLED_DAYS - self.lag) if self.lag is not None else 0 def quiet_days(rows): # About 7 expected rows' worth of silence, at least 2 days, padded while the window is still filling. return max(2, int(math.ceil(7.0 * self.days / rows))) + pad rows_by = {s: n for s, n in self.inv.by_source(tool, names).items() if n > 0} last_by = self.inv.last_by_source(tool, names) if any(not last_by.get(s) for s in rows_by): self.no_dates[tool].append(label) # printed as NOT CHECKED, and the D01 PASS claims no stop check if not last_by: return sev = "FAIL" if self.tier(e) == 1 or e["status"] == "verified" else "WARN" disp = ("fix implementation; verified becomes implemented until the acceptance tests pass again" if e["status"] == "verified" else "fix implementation") if all(last_by.get(s) for s in rows_by): last_all, gap = max(last_by.values()), quiet_days(total) if last_all < self.we - timedelta(days=gap): self.add("D01", sev, tool, e["name"], "", f"{label}: stopped inside the window. Last seen {last_all}, then no rows for " f"{(self.we - last_all).days} days to {self.we}; at {num(total)} rows in {self.days} days a " f"silence over {gap} days is not chance. Status {e['status']}.", disp, affected=total, unit="rows before it stopped", last_seen=last_all) return expected = self.expected_sources(e, tool) live = [s for s in rows_by if last_by.get(s) and last_by[s] >= self.we - timedelta(days=quiet_days(rows_by[s]))] for s in sorted(rows_by): if not s or s == "unknown" or (expected and s not in expected) or s in live or not last_by.get(s): continue others = sorted(x or "no source label" for x in live) verb = "sends" if len(others) == 1 else "send" self.add("D01", sev, tool, e["name"], "", f"{label}: stopped from {s} inside the window. Last seen {last_by[s]} after " f"{num(rows_by[s])} rows, while {', '.join(others) or 'no other source'} still {verb}. " f"Status {e['status']}.", disp, source=s, affected=rows_by[s], denominator=total, unit="rows from the source before it stopped", last_seen=last_by[s]) def check_sources(self, tool, e, label, names, total): """D06: sources the plan does not list, and planned sources with no rows.""" expected = self.expected_sources(e, tool) if not expected: return seen = {s: n for s, n in self.inv.by_source(tool, names).items() if n > 0} labelled = {s: n for s, n in seen.items() if s and s != "unknown"} complete = sum(labelled.values()) >= total if not labelled or not complete: self.unlabelled[tool].append(label) if not labelled: return self.checked["D06"] += 1 accept, skip_missing = set(expected), set() if tool == "ga4" and "server" in expected: # GA4 records the data stream platform, so Measurement Protocol and server-side rows arrive as a stream. accept |= {"web", "ios", "android"} skip_missing.add("server") self.stream_platform[tool].append(label) sev = "FAIL" if self.tier(e) == 1 else "WARN" for s in sorted(x for x in labelled if x not in accept): self.add("D06", sev, tool, e["name"], "", f"{label}: {num(labelled[s])} of {num(total)} rows come from {s}, which the plan does not list " f"for {tool} (planned: {', '.join(sorted(expected))}). An unexpected source is usually a second " "producer.", "fix implementation", source=s, affected=labelled[s], denominator=total, unit="rows from the unplanned source") missing = sorted(s for s in expected if s not in labelled and s not in skip_missing) if complete else [] if missing: verb = "sends" if len(labelled) == 1 else "send" self.add("D06", sev, tool, e["name"], "", f"{label}: no rows from {', '.join(missing)}, which the plan lists, while " f"{', '.join(sorted(labelled))} {verb} {num(total)}.", "fix implementation", source=",".join(missing), affected=0, unit="rows from the planned source") def check_duplicates(self, tool, e, label, names, total): """D12: repeated occurrence keys, and transport duplicates as INFO.""" rows = [o for n in names for o in self.inv.events[tool].get(n, [])] if any(o["duplicate_keys"] is not None for o in rows): self.checked["D12"] += 1 self.checked["dup_keys"] += 1 dup = sum(o["duplicate_keys"] or 0 for o in rows) tdup = sum(o["transport_duplicates"] or 0 for o in rows) if dup: self.add("D12", "FAIL" if self.tier(e) == 1 else "WARN", tool, e["name"], "", f"{label}: {plural(dup, 'repeated occurrence key')} in {num(total)} rows; a repeated business " "occurrence counts twice wherever nothing deduplicates (GA4 export rows and warehouse models " "keep both).", "fix implementation", affected=dup, denominator=total, unit="repeated keys") if tdup: self.add("D12", "INFO", tool, e["name"], "", f"{label}: {plural(tdup, 'transport duplicate')} (same insert or message ID); " f"{TRANSPORT_NOTES.get(tool, DEFAULT_TRANSPORT_NOTE)}.", affected=tdup, denominator=total, unit="transport duplicates") def check_unplanned(self, tool, cls, notes, planned): """D02 orphans, D03 spelling variants, D04 dynamic and per-page names.""" inv = self.inv rank, ranked = self.ranks(tool) tool_rows = inv.tool_total(tool) groups, generated = defaultdict(list), [] for n, c in sorted(cls.items()): total = inv.total(tool, [n]) last = inv.last_seen(tool, [n]) if c == "dynamic": groups[notes[n]].append(n) elif c == "generated": generated.append(n) elif c in ("orphan", "test"): disp = ("retire after confirming no real users send it" if c == "test" else "needs decision (adopt into the plan or retire)") note = f" It {notes[n]}." if n in notes else "" if not inv.has_counts(tool, n): what = (f"listed by {tool} without counts" if inv.events[tool].get(n) else "seen as a table or column list only") self.add("D02", "WARN", tool, n, "", f"{n}: {what}, not in the plan; volume unknown, so it is not " f"ranked.{note}", disp) continue if total == 0: self.add("D02", "INFO", tool, n, "", f"{n}: unplanned, seen only before the window (last seen " f"{last or 'unknown'}).", last_seen=last) continue share = total / tool_rows if tool_rows else 0 tags = [f"rank {rank[n]} of {ranked} names by volume, {pct(share)} of {tool} rows" + (", priority" if rank[n] <= 20 and share >= 0.01 else "")] if c == "test": tags.append("looks like test traffic") self.add("D02", "WARN", tool, n, "", f"{n}: {num(total)} rows in the window, not in the plan ({'; '.join(tags)}).{note}", disp, affected=total, denominator=tool_rows, unit="rows", last_seen=last) elif c == "variant": self.report_variant(tool, n, notes[n], total, last, planned) for pattern, members in sorted(groups.items()): rows = inv.total(tool, members) self.add("D04", "FAIL", tool, pattern, "", f"{len(members)} event names follow {pattern} (for example {masked_example(members[0])}, " f"{masked_example(members[-1])}), {num(rows)} rows.", "fix implementation (move the variable part into a property)", affected=len(members), unit="event names") if generated: rows = inv.total(tool, generated) sev = "FAIL" if len(generated) >= self.args.dynamic_min else "WARN" examples = ", ".join(masked_example(n) for n in generated[:2]) self.add("D04", sev, tool, "(per-page names)", "", f"{plural(len(generated), 'event name')} follow 'Viewed or Loaded Page or Screen' (for " f"example {examples}), {num(rows)} rows: a CDP destination's Track Named Pages or Track " "Categorized Pages setting creates one event type per page.", "fix the destination setting (Track All Pages sends one Loaded a Page event with the page name " "as a property), after moving the reports that read the old names", affected=len(generated), unit="event names") def report_variant(self, tool, n, other, total, last, planned): inv = self.inv other_total = inv.total(tool, [other]) def srcs(x): return src_list(inv.by_source(tool, [x])) or "no source" if other in planned and total > 0: self.add("D03", "FAIL", tool, n, "", f"'{n}' ({srcs(n)}, {num(total)} rows) beside planned '{other}' ({srcs(other)}, " f"{num(other_total)} rows): one action under two spellings, " f"{pct(total / (total + other_total))} of its rows under the unplanned one.", "fix implementation (stop or rename the non-canonical emission; tool-side merges and display " "names leave raw data split)", affected=total, denominator=total + other_total, unit="rows under the variant", last_seen=last) elif other in planned: seen = last or ("unknown, listed without counts" if not inv.has_counts(tool, n) else "unknown") self.add("D03", "WARN", tool, n, "", f"stale variant '{n}' of planned '{other}', last seen {seen}; check dashboards that still filter " "on it.", "needs decision (map the history or retire the name)", affected=0, unit="rows under the variant", last_seen=last) elif n < other: # report an unplanned pair once both = total > 0 and other_total > 0 self.add("D03", "FAIL" if both else "WARN", tool, n, "", f"unplanned spellings '{n}' ({num(total)} rows) and '{other}' ({num(other_total)} rows) of one " "action.", "needs decision (adopt one spelling into the plan, stop the other)", affected=total, denominator=(total + other_total) or None, unit="rows under the first spelling", last_seen=last) # ---------- report and --out writer ---------- def report(self): a = self.args label, why = self.verdict if a.json: print(json.dumps({ "window": [str(self.ws) if self.ws else None, str(self.we) if self.we else None], "as_of": str(self.as_of), "provisional": self.provisional, "tools": self.tools, "verdict": label, "verdict_reason": why, "blocking_rules": self.blocking, "findings": self.findings, "classes": {t: dict(c[0]) for t, c in self.classes.items()}, "matrix": self.matrix, "summary": summary_counts(self.findings), }, indent=2, default=str)) return print(f"Tracking plan audit, drift: {len(self.plan.events)} planned events, {len(self.plan.props)} property " f"rows; inventories for {', '.join(self.tools) or 'no tools'} ({plural(len(self.inv.files), 'file')})") as_of = f"as of {self.as_of}" + ("" if self.as_of_given else " (default: today, or the window end plus 4 days when that is earlier)") print(f"Window {self.ws or '?'} to {self.we or '?'}, {as_of}. Thresholds: type " f"{pct(a.mismatch_share)}, required {pct(a.required_fail)}, sample {a.min_sample}, global " f"{pct(a.global_share)}, dynamic {a.dynamic_min}, optional {num(a.optional_min_rows)} rows, GA4 " f"parameters {a.ga4_param_limit}.") if self.provisional: print(f"The window {self.lag_text}. GA4 keeps filling daily tables for up to {LATE_DAYS} days and mobile SDKs " "deliver late, so the last days are provisional.") print("Plan checks: " + "; ".join(f"{r['id']} {r['severity']}" for r in self.blocking) + f". Verdict: {label} ({why}).") for r in self.blocking: if r["severity"] != "PASS": print(f" {r['id']} {r['severity']}: {r['message']}") derived = [f"{e['name']} ({', '.join(e['derived'])})" for e in self.plan.events.values() if e["derived"]] if derived: print(f"Browser companions read from delivery_paths as web sources: {first_items(derived, 4)}.") untiered = [e["name"] for e in self.plan.events.values() if e["tier"] is None] if untiered: print(f"No risk tier on {len(untiered)} events ({first_items(untiered, 4)}); tier 1 thresholds were not " "applied to them.") print("") for f in self.findings: name = f"{f['rule']} {f['name']}" line = f"{f['severity']} {name}" + (f" {f['tool']}" if f["tool"] else "") + f": {f['message']}" if f["default_disposition"] and f["severity"] in ("FAIL", "WARN"): line += f" Default disposition: {f['default_disposition']}." print(line) print("\nObserved names by class:") for tool, (cls, notes) in self.classes.items(): counts = OrderedDict((c, 0) for c in CLASSES) for c in cls.values(): counts[c] += 1 print(f" {tool}: " + ", ".join(f"{n} {c}" for c, n in counts.items()) + f" ({len(cls)} names)") if self.matrix: print("\nCross-tool matrix (root-cause locator):") for line in self.matrix: print(f" {line}") print("\n" + summary_counts(self.findings)) def write_out(self): out = self.args.out target = os.path.realpath(out) # symlinks resolved, so a link cannot lead into either folder if "martech-context" in target.split(os.sep): fail("--out points inside martech-context/; write drafts to a scratch folder and propose changes for " "approval") skill = os.path.realpath(SKILL_DIR) if target == skill or target.startswith(skill + os.sep): fail("--out points inside the skill folder; write audit outputs to a scratch folder") os.makedirs(out, exist_ok=True) window = f"{self.ws or ''} to {self.we or ''}".strip() listed = [f for f in self.findings if f.get("id")] with open(os.path.join(out, "findings.csv"), "w", newline="", encoding="utf-8") as handle: w = csv.writer(handle) w.writerow(["finding_id", "rule", "severity", "tool", "source", "event", "property", "affected", "unit", "denominator", "share_pct", "last_seen", "window", "message", "default_disposition"]) for f in listed: w.writerow([f["id"], f["rule"], f["severity"], f["tool"], f["source"], f["event"], f["property"], f["affected"] if f["affected"] is not None else "", f["unit"], f["denominator"] or "", f["share_pct"] if f["share_pct"] is not None else "", f["last_seen"], window, f["message"], f["default_disposition"]]) with open(os.path.join(out, "observed-snapshot.csv"), "w", newline="", encoding="utf-8") as handle: w = csv.writer(handle) w.writerow(["event", "tool", "tool_name", "status", "risk_tier", "planned_sources", "sources_seen", "rows_in_window", "last_seen", "finding_ids"]) for tool in self.classes: for e in self.plan.events.values(): fn = self.resolved[tool][2][e["name"]][0] if not fn: continue names = self.obs_names(tool, fn) names_only = (not any(self.inv.has_counts(tool, n) for n in names) and any(self.inv.listed(tool, n) for n in names)) ids = [f["id"] for f in listed if f["tool"] == tool and f["event"] == e["name"]] w.writerow([e["name"], tool, fn, e["status"], e["tier"] or "", ";".join(sorted(self.expected_sources(e, tool))), ";".join(sorted(s or "no source label" for s in self.inv.by_source(tool, names))), "" if names_only else self.inv.total(tool, names), self.inv.last_seen(tool, names) or "", ";".join(ids)]) self.write_plan_changes(out, listed) print(f"\nWrote findings.csv, observed-snapshot.csv and plan-changes-draft.csv to {out} (drafts, nothing " "applied).") def write_plan_changes(self, out, listed): approver = {1: "business, data and implementation owners", 2: "data and implementation owners", 3: "data owner"} with open(os.path.join(out, "plan-changes-draft.csv"), "w", newline="", encoding="utf-8") as handle: w = csv.writer(handle) w.writerow(["change_id", "finding_ids", "event", "property", "field", "current_value", "proposed_value", "disposition", "change_class", "version_bump", "reason", "approver_role", "status"]) n = 0 for f in listed: e = self.plan.events.get(f["event"]) tier = e["tier"] if e else 2 row = None if f["rule"] == "D01" and f["severity"] in ("FAIL", "WARN") and e: if "Root cause: the planned GA4 name" in f["message"]: row = ["ga4_name", e["row"].get("ga4_name", "")] + self.ga4_name_change(f["message"]) elif "stopped" in f["message"] and e["status"] == "verified": row = ["status", "verified", "implemented", "fix implementation", "documentation", "no", f"stopped in {f['tool']}; verified again after the acceptance tests pass"] elif "stopped" not in f["message"]: row = ["status", e["status"], "decide: keep (fix the emitter) or approved", "needs decision", "documentation", "no", f"never seen in {f['tool']}"] elif f["rule"] == "D05" and e: if e["status"] in ("proposed", "approved"): row = ["status", e["status"], "implemented", "amend plan", "documentation", "no", f"already collected in {f['tool']}"] elif e["status"] == "deprecated": row = ["status_changed_at", e["row"].get("status_changed_at", ""), "decide: confirm the end date (version tail), then retire once emission stops", "needs decision", "documentation", "no", f"deprecated and still sent to {f['tool']}"] else: row = ["status", e["status"], "decide: deprecated with an end date, or fix the emitter", "needs decision", "documentation", "no", f"{e['status']} but still sent to {f['tool']}"] elif f["rule"] == "D02" and f["severity"] == "WARN": row = ["event", "not in plan", "decide: adopt (write a contract) or retire", "needs decision", "compatible if adopted", "no", f"unplanned in {f['tool']}"] elif f["rule"] in ("D08", "D09") and f["severity"] == "WARN": current = "optional, never sent" if f["rule"] == "D08" else "not in plan" proposed = "decide: implement or remove" if f["rule"] == "D08" else "decide: adopt or stop sending" if f["rule"] == "D08": reason = "optional property never sent in " else: reason = "unplanned global property in " if f["event"] == "*" else "unplanned property in " row = ["property", current, proposed, "needs decision", "compatible if adopted" if f["rule"] == "D09" else "documentation", "no", reason + f["tool"]] elif f["rule"] == "D11" and f["severity"] in ("FAIL", "WARN") and e: p = next((q for q in self.plan.props_for(e["name"]) if q["name"] == f["property"]), None) values = [v for v in re.findall(r"'([^']+)' [\d,]+", f["message"]) if v not in REDACTION_TOKENS] row = ["allowed_values", "|".join(p["allowed"]) if p else "", "decide: add " + "|".join(values) if values else "decide", "needs decision", "compatible", "no", "new values observed; update Protocols, Amplitude Data, Avo or RudderStack copies first"] elif f["rule"] == "D14" and "planned GA4 name" in f["message"]: row = ["ga4_name", e["row"].get("ga4_name", "") if e else "", "decide: a name of 40 characters or fewer", "needs decision", "rename", "yes", "over the GA4 name limit"] if row: n += 1 w.writerow([f"PC-{n}", f["id"], f["event"], f["property"]] + row + [approver.get(tier, approver[2]), "draft"]) @staticmethod def ga4_name_change(message): """Proposed value through reason for a D01 finding whose root cause is the planned GA4 name.""" needs, reasons = [], [] length = "is 40 characters or fewer" if "GA4 does not log event names over 40" in message: needs.append(length) reasons.append("GA4 does not log event names over 40 characters") if "breaks GA4's event naming rule" in message: needs.append("starts with a letter and uses only letters, numbers and underscores") reasons.append("the name breaks GA4's event naming rule") if "GA4 reserves" in message: needs.append("is not reserved by GA4") reasons.append("GA4 reserves the name or its prefix") proposed = ("decide: a name of 40 characters or fewer" if needs == [length] else "decide: a GA4 name that " + " and ".join(needs)) return [proposed, "needs decision", "rename", "yes", "; ".join(reasons)] def cmd_drift(args): drift = Drift(args) drift.run() drift.report() if args.out: drift.write_out() return 1 if any(f["severity"] == "FAIL" for f in drift.findings) else 0