#!/usr/bin/env python3 """Mechanical checks over a tracking plan kept as two CSV tabs. Usage: python3 check_plan.py EVENTS.csv PROPERTIES.csv [--max-name 40] [--reserved FILE] [--ga4-tag-params N] EVENTS.csv and PROPERTIES.csv use the columns of assets/plan-sheet-events.csv and assets/plan-sheet-properties.csv. Standard library only. Prints one line per problem and exits with status 1 when there is at least one. Checks: - final names per destination after lowercasing and replacing spaces with underscores: two plan rows that end up as one name in one destination - one destination reached twice for one event in delivery_paths (two producers of one final name) - event and property names longer than --max-name (default 40, the GA4 limit) - names equal to an entry in the --reserved file (one name per line; take the current lists from the vendor docs, GA4 web and app lists differ) - one type per property name across events - property and HubSpot names that collide once lowercased, and HubSpot names that break its internal-name rule (lowercase letters, digits and underscores, starting with a letter, 50 characters at most) - GA4 parameters per event, plus --ga4-tag-params for parameters the tag adds (count them from one DebugView capture) It cannot check meaning, triggers, consent or identity. Those stay in the acceptance tests. """ import argparse import csv import re import sys from collections import defaultdict DESTINATION_COLUMNS = { "ga4": "ga4_name", "mixpanel": "mixpanel_name", "amplitude": "amplitude_name", "segment": "segment_name", "rudderstack": "rudderstack_name", } NOT_MAPPED = {"", "none", "not sent", "n/a", "-"} GA4_PARAMETER_LIMIT = 25 HUBSPOT_NAME = re.compile(r"^[a-z][a-z0-9_]{0,49}$") HUBSPOT_TOKEN = re.compile(r"\b[A-Za-z][A-Za-z0-9]*_[A-Za-z0-9_]+\b") def normalise(name): """Lowercase and replace runs of whitespace with one underscore.""" return re.sub(r"\s+", "_", name.strip()).lower() def read_rows(path): with open(path, newline="", encoding="utf-8") as handle: return list(csv.DictReader(handle)) def cell(row, column): return (row.get(column) or "").strip() def main(): parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument("events") parser.add_argument("properties") parser.add_argument("--max-name", type=int, default=40) parser.add_argument("--reserved", help="file with one reserved name per line") parser.add_argument("--ga4-tag-params", type=int, default=None) args = parser.parse_args() events = read_rows(args.events) properties = read_rows(args.properties) problems = [] reserved = set() if args.reserved: with open(args.reserved, encoding="utf-8") as handle: for line in handle: if line.strip() and not line.startswith("#"): reserved.add(normalise(line)) # Event names per destination. producers = defaultdict(list) for row in events: event = cell(row, "event") or "?" for destination, column in DESTINATION_COLUMNS.items(): name = cell(row, column) if name.lower() in NOT_MAPPED: continue final = normalise(name) producers[(destination, final)].append(event) if len(final) > args.max_name: problems.append(f"{destination}: {final} is {len(final)} characters, over {args.max_name}") if final in reserved: problems.append(f"{destination}: {final} is on the reserved list") for (destination, final), sources in sorted(producers.items()): if len(sources) > 1: problems.append( f"{destination}: final name {final} comes from {', '.join(sources)}; " "rename one or write a reconciliation rule" ) # Two delivery paths to one destination for one event. for row in events: paths = defaultdict(list) for part in cell(row, "delivery_paths").split(";"): bits = part.split(" via ") if len(bits) == 2: paths[bits[0].strip().lower()].append(bits[1].strip()) for destination, vias in paths.items(): if len(vias) > 1: problems.append( f"{cell(row, 'event')}: {destination} is reached via {', '.join(vias)}; " "two producers of one final name" ) # Property names, types and GA4 parameters. types = defaultdict(set) spellings = defaultdict(set) ga4_parameters = defaultdict(set) for row in properties: prop = cell(row, "property") if not prop: continue types[prop].add(cell(row, "type").lower()) spellings[normalise(prop)].add(prop) if len(prop) > args.max_name: problems.append(f"property {prop} is {len(prop)} characters, over {args.max_name}") if normalise(prop) in reserved: problems.append(f"property {prop} is on the reserved list") parameter = cell(row, "ga4_parameter") if parameter: ga4_parameters[cell(row, "event") or "?"].add(parameter) for prop, found in sorted(types.items()): if len(found) > 1: problems.append(f"property {prop} has more than one type: {', '.join(sorted(found))}") for key, names in sorted(spellings.items()): if len(names) > 1: problems.append(f"properties {', '.join(sorted(names))} collide as {key}") # HubSpot names mentioned in hubspot_route. hubspot = defaultdict(set) for row in events: for token in HUBSPOT_TOKEN.findall(cell(row, "hubspot_route")): hubspot[token.lower()].add(token) if not HUBSPOT_NAME.match(token): problems.append(f"hubspot: {token} breaks the internal-name rule") for key, names in sorted(hubspot.items()): if len(names) > 1: problems.append(f"hubspot: {', '.join(sorted(names))} become one property, {key}") # GA4 parameter count. Rows with event "*" apply to every event. shared = ga4_parameters.pop("*", set()) for row in events: if cell(row, "ga4_name").lower() in NOT_MAPPED: continue event = cell(row, "event") count = len(ga4_parameters.get(event, set()) | shared) if args.ga4_tag_params is None: print(f"note: ga4 {event} has {count} plan parameters; tag-added parameters not counted (pass --ga4-tag-params)") else: count += args.ga4_tag_params if count > GA4_PARAMETER_LIMIT: problems.append(f"ga4: {event} sends {count} parameters, over {GA4_PARAMETER_LIMIT}") for problem in problems: print(f"problem: {problem}") print(f"{len(problems)} problem(s) in {len(events)} events and {len(properties)} property rows") return 1 if problems else 0 if __name__ == "__main__": sys.exit(main())