"""Drift rule D12 against system-of-record counts, and the cross-tool matrix. A mixin for the Drift class in drift.py. Thresholds and reasons: references/drift-rules.md. """ from collections import defaultdict from common import LATE_DAYS, Plan, fail, first_items, norm_route, num, parse_date, pct, ratio_text, to_int class CountRules(object): # ---------- reference counts, D12 ---------- def check_reference(self, audited): inv = self.inv if self.reference is None: tier1 = [e["name"] for e in self.plan.events.values() if e["tier"] == 1 and e["status"] in ("implemented", "verified")] if tier1: self.add("D12", "NOT CHECKED", "", "", "", f"no reference file; tier 1 events need system-of-record counts ({first_items(tier1, 4)}).") return covered = set() for row in self.reference: tool = row.get("tool", "").lower() e = self.plan.events.get(row["event"]) if tool not in audited or e is None: self.add("D12", "NOT CHECKED", tool, row["event"], "", f"reference row for {row['event']} in {tool or 'no tool'} has no matching planned event or " "inventory.") continue covered.add((row["event"], tool)) fn = self.plan.final_name(e, tool)[0] or row["event"] names = self.obs_names(tool, fn) count = to_int(row.get("count"), self.args.reference, row, "count") if not count: self.add("D12", "NOT CHECKED", tool, e["name"], "", f"{fn}: reference count is empty or zero, so no " "ratio.") continue if not any(inv.has_counts(tool, n) for n in names) and any(inv.listed(tool, n) for n in names): self.add("D12", "NOT CHECKED", tool, e["name"], "", f"{fn}: listed by {tool} without counts, so no " "ratio.") continue observed = inv.total(tool, names) ratio = observed / count self.ratios[e["name"]][tool] = ratio route = norm_route(row.get("route")) if route is None: raw = row.get("route", "") what = f"route '{raw}' is neither server nor browser" if raw else "the route cell is empty" self.add("D12", "NOT CHECKED", tool, e["name"], "", f"{fn}: ratio {ratio_text(ratio)} was not judged, because {what}. Set server or browser in the " "reference file; the route decides which threshold applies.") continue self.checked["D12"] += 1 self.checked["ratios"] += 1 self.judge_ratio(row, tool, e, fn, route, observed, count, ratio) missing = sorted({(e["name"], t) for e in self.plan.events.values() for t in audited if e["tier"] == 1 and e["status"] in ("implemented", "verified") and self.plan.final_name(e, t)[0]} - covered) if missing: self.add("D12", "NOT CHECKED", "", "", "", f"no reference count for tier 1 events: {first_items([f'{n} ({t})' for n, t in missing], 4)}.") if self.tz_missing: self.add("D12", "NOT CHECKED", "", "", "", f"no time zone on both sides for {first_items(self.tz_missing, 4)}; GA4 event_date follows the " "property time zone and event_timestamp is UTC, so fill timezone and tool_timezone before " "trusting a ratio near 1.0.") def judge_ratio(self, row, tool, e, fn, route, observed, count, ratio): try: low = float(row["band_low"]) if row.get("band_low") else None high = float(row["band_high"]) if row.get("band_high") else None except ValueError: fail(f"{self.args.reference} line {row['_line']}: band_low and band_high must be numbers such as 0.85") tz_ref, tz_tool = row.get("timezone", "").strip(), row.get("tool_timezone", "").strip() aligned = bool(tz_ref and tz_tool and tz_ref.lower() == tz_tool.lower()) if not (tz_ref and tz_tool): self.tz_missing.append(f"{fn} ({tool})") if aligned: tz = f"days cut in {tz_ref} on both sides" elif tz_ref or tz_tool: tz = f"reference in {tz_ref or 'an unrecorded time zone'}, {tool} in {tz_tool or 'an unrecorded time zone'}" else: tz = "time zones not recorded" sot = row.get("source_of_truth", "") or "the system of record" base = (f"{fn}: {num(observed)} events against {num(count)} from {sot}, ratio {ratio_text(ratio)}, " f"{route} route, {tz}") window = "" rs, re_ = parse_date(row.get("window_start", "")), parse_date(row.get("window_end", "")) if (rs and self.ws and rs != self.ws) or (re_ and self.we and re_ != self.we): window = f" The reference window ({rs} to {re_}) differs from the audit window." kw = {"affected": observed, "denominator": count, "unit": "events in the tool"} if not aligned and 0.99 <= ratio <= 1.01: why = "the time zones differ" if (tz_ref and tz_tool) else "a time zone is not recorded" self.add("D12", "WARN", tool, e["name"], "", f"{base}: within 1% of the reference while {why}; a day-boundary offset can explain a gap this " f"small, so cut both sides in one time zone and run again before acting.{window}", "needs decision (align the time zones, then judge)", **kw) elif ratio > 1.0: if high is not None and high > 1.0 and ratio <= high: self.add("D12", "WARN", tool, e["name"], "", f"{base}: above 1.0 but inside the documented band; " f"check the definition.{window}", "needs decision", **kw) else: band = f" (documented band up to {high:.2f})" if high is not None else "" self.add("D12", "FAIL", tool, e["name"], "", f"{base}: more events than facts, which only duplicates, a second producer or a different " f"definition explain{band}.{window}", "fix implementation", **kw) elif route == "server" and ratio < 0.99: if self.provisional: self.add("D12", "WARN", tool, e["name"], "", f"{base}: below 0.99 on a server route, but the window {self.lag_text} and GA4 keeps filling " f"daily tables for up to {LATE_DAYS} days; run again after that before treating it as loss." f"{window}", "needs decision (run again after the refresh window)", **kw) else: self.add("D12", "FAIL", tool, e["name"], "", f"{base}: a server route loses nothing to consent or blockers, so below 0.99 fails once the " f"refresh window has passed.{window}", "fix implementation", **kw) elif route != "server" and ((low is not None and ratio < low) or (high is not None and ratio > high)): self.add("D12", "WARN", tool, e["name"], "", f"{base}: outside the documented band ({low if low is not None else '-'} to " f"{high if high is not None else '-'}).{window}", "fix implementation", **kw) elif route != "server" and low is None and high is None: self.no_band += 1 # reported, not judged, so the D12 PASS must not call it a match self.add("D12", "INFO", tool, e["name"], "", f"{base}; no band documented, so ask the contract owner for " f"one.{window}", **kw) else: self.add("D12", "INFO", tool, e["name"], "", f"{base}; within the expected range.{window}", **kw) # ---------- cross-tool matrix ---------- def check_matrix(self, audited): if len(audited) < 2: return for name, ratios in self.ratios.items(): if len(ratios) > 1: above = [t for t, r in ratios.items() if r > 1.0] text = ", ".join(f"{t} {ratio_text(r)}" for t, r in sorted(ratios.items())) hint = ("Above the reference in every tool: start at the emitter or the definition." if len(above) == len(ratios) else f"Above the reference only in {', '.join(above)}: start at that tool's path." if above else "No tool above the reference.") self.matrix.append(f"{name}: ratio to the reference {text}. {hint}") flagged = defaultdict(set) for f in self.findings: if f["rule"] in ("D07", "D10", "D11") and f["severity"] in ("FAIL", "WARN"): flagged[(f["event"], f["property"])].add(f["tool"]) for e in self.plan.events.values(): for p in self.plan.props_for(e["name"]): key = (e["name"], p["name"]) if key not in flagged: continue cells, tools = [], [] for tool in audited: fn = self.resolved[tool][2].get(e["name"], (None, False))[0] if not fn or tool in p["excluded"]: continue names = self.obs_names(tool, fn) pn = Plan.tool_prop_name(p, tool) den = sum(self.inv.by_source(tool, names).values()) every = [o for n in names for o in self.inv.props[tool].get(n, {}).get(pn, [])] rows = [o for o in every if o["rows"] is not None] if every and not rows: # warehouse or Lexicon column: a type, no counts types = "/".join(sorted({o["type"] for o in every if o["type"]})) or "untyped" cells.append(f"{tool} column {types}, no counts") tools.append(tool) continue present = max(sum(o["rows"] for o in rows if o["type"] != "null") - sum((o["sentinel_hits"] or 0) + (o["empty_hits"] or 0) for o in rows), 0) types = sorted({o["type"] for o in rows if o["type"] and o["type"] != "null"}) if den: cells.append(f"{tool} {pct(present / den)} present, {'/'.join(types) or 'no rows'}") tools.append(tool) if len(tools) < 2: continue bad = flagged[key] & set(tools) hint = ("Flagged in every tool: compare the emitter first." if bad == set(tools) else f"Flagged only in {', '.join(sorted(bad))}: start at that tool's path (tag, CDP mapping, " "connector, transformation or event rule).") ga4 = f" (ga4: {p['ga4']})" if p["ga4"] and p["ga4"] != p["name"] else "" self.matrix.append(f"{e['name']}.{p['name']}{ga4}: {'; '.join(cells)}. {hint}")