#!/usr/bin/env python3 """Build tagged URLs from a UTM registry CSV. Standard library only. Usage: python3 build_links.py --registry utm-registry.csv --pairs allowed-pairs.csv \ --hosts example.com,www.example.com --out utm-registry-built.csv Builds rows whose status is "approved" and that have no param_template. Checks each row against the allowed pairs file (pair, ambiguity, content slots). Writes every input row to a new file with tagged_url and build_result. Exits with status 1 if any built row has a problem. Never edits the registry. """ import argparse import csv import os import re import sys from urllib.parse import parse_qsl, quote, urlencode, urlsplit, urlunsplit ORDER = ( "utm_source", "utm_medium", "utm_campaign", "utm_id", "utm_source_platform", "utm_content", "utm_term", "utm_creative_format", "utm_marketing_tactic", ) REQUIRED = ("utm_source", "utm_medium", "utm_campaign", "utm_id") SLUG = re.compile(r"^[a-z0-9_]+$") MEDIUM = re.compile(r"^[a-z0-9_-]+$") CAMPAIGN = re.compile(r"^20[0-9]{2}_(0[1-9]|1[0-2])_[a-z0-9_]+$") ID = re.compile(r"^[A-Za-z0-9_-]+$") DIGIT_RUN = re.compile(r"[0-9]{7,}") # phone numbers and personal IDs CAMPAIGN_MAX = 60 WARN_LENGTH = 800 # leaves room for click IDs added at click time MAX_LENGTH = 1000 # GA4 page_location limit, counted on the encoded URL def load_pairs(path): """Read allowed-pairs.csv into {(source, medium): rules}.""" pairs = {} with open(path, newline="", encoding="utf-8-sig") as handle: for row in csv.DictReader(handle): source = (row.get("utm_source") or "").strip() medium = (row.get("utm_medium") or "").strip() if not source or not medium: continue slots = { slot.strip() for slot in (row.get("content_slots") or "").split("|") if slot.strip() } pairs[(source, medium)] = { "ambiguous": (row.get("ambiguous") or "").strip().lower() == "yes", "passed": (row.get("test_result") or "").strip().lower().startswith("pass"), "slots": slots, } return pairs def problems_for(values, landing, row, pairs, hosts): found = [] for key in REQUIRED: if not values[key]: found.append(f"missing {key}") for key, value in values.items(): if not value: continue if "{" in value or "}" in value: found.append(f"{key} holds a macro; enter it in the platform instead") elif "@" in value: found.append(f"{key} contains @ and may be personal data") elif DIGIT_RUN.search(value): found.append(f"{key} has 7 or more digits in a row; check it is not a phone number or personal ID") elif key == "utm_id": if not ID.match(value): found.append("utm_id has characters outside A-Z, a-z, 0-9, _ and -") elif key == "utm_medium": if not MEDIUM.match(value): found.append("utm_medium is not lowercase a-z, 0-9, _ and -") elif key == "utm_campaign": if not CAMPAIGN.match(value) or len(value) > CAMPAIGN_MAX: found.append("utm_campaign is not yyyy_mm_initiative (month 01 to 12) within 60 characters") elif not SLUG.match(value): found.append(f"{key} is not lowercase a-z, 0-9 and _") pair = pairs.get((values["utm_source"], values["utm_medium"])) if pair is None: found.append("source/medium pair is not in the allowed pairs file") else: if pair["ambiguous"] and not pair["passed"]: found.append("pair matches rules for two channels and has no passing test visit") content = values["utm_content"] if pair["slots"] and content and content not in pair["slots"]: found.append("utm_content is not a slot listed for this pair") if not (row.get("expected_channel") or "").strip(): found.append("expected_channel is empty; predict the channel before approval") if landing.scheme != "https" or (landing.hostname or "") not in hosts: found.append("landing_url is not https on an approved landing host") return found def build(values, landing): kept = [ (key, value) for key, value in parse_qsl(landing.query, keep_blank_values=True) if not key.lower().startswith("utm_") ] tags = [(key, values[key]) for key in ORDER if values[key]] query = urlencode(kept + tags, quote_via=quote, safe="") return urlunsplit((landing.scheme, landing.netloc, landing.path, query, landing.fragment)) def main(argv=None): parser = argparse.ArgumentParser(description="Build tagged URLs from a UTM registry CSV.") parser.add_argument("--registry", required=True, help="registry CSV, for example utm-registry.csv") parser.add_argument("--pairs", required=True, help="allowed pairs CSV, for example allowed-pairs.csv") parser.add_argument("--hosts", required=True, help="comma-separated landing hosts, for example example.com,www.example.com") parser.add_argument("--out", required=True, help="new output CSV; never the registry itself") args = parser.parse_args(argv) if os.path.abspath(args.out) == os.path.abspath(args.registry): sys.exit("write to a new file; the registry changes only after the user approves") hosts = {host.strip().lower() for host in args.hosts.split(",") if host.strip()} pairs = load_pairs(args.pairs) with open(args.registry, newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) fieldnames = list(reader.fieldnames or []) rows = list(reader) for column in ("tagged_url", "build_result"): if column not in fieldnames: fieldnames.append(column) seen = {} failed = 0 for line, row in enumerate(rows, start=2): if (row.get("status") or "").strip() != "approved": row["build_result"] = "skipped: status is not approved" continue if (row.get("param_template") or "").strip(): row["build_result"] = "skipped: platform template row" continue values = {key: (row.get(key) or "").strip() for key in ORDER} landing = urlsplit((row.get("landing_url") or "").strip()) found = problems_for(values, landing, row, pairs, hosts) tagged = "" if found else build(values, landing) if tagged and len(tagged) > MAX_LENGTH: found.append(f"tagged URL is {len(tagged)} characters, over {MAX_LENGTH}") if tagged and tagged in seen: found.append(f"same tagged URL as line {seen[tagged]}") if found: failed += 1 row["tagged_url"] = "" row["build_result"] = "; ".join(found) continue seen[tagged] = line row["tagged_url"] = tagged row["build_result"] = "ok" if len(tagged) > WARN_LENGTH: row["build_result"] = f"ok, but {len(tagged)} characters" with open(args.out, "w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) print(f"{len(rows)} rows read, {failed} with problems; output in {args.out}") return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())