`tracking-coverage-matrix.py`

112 lines
#!/usr/bin/env python3
"""Per-site tracking-coverage matrix — $0 local, no LLM.
Gathers owned domains (Cloudflare + GoDaddy + GA4 slugs), curls each in parallel,
detects GTM / GA4 / Meta Pixel / Pinterest in the HTML, cross-refs GSC grants from
the ga-allsites cache, and prints a matrix + punch-lists for the LIVE sites.
"""
import json
import os
import pathlib
import re
import ssl
import urllib.request
from concurrent.futures import ThreadPoolExecutor

HERE = pathlib.Path(__file__).parent
SEC = pathlib.Path.home() / "Projects" / "secrets-manager" / ".env"

# Probe tuning — tweak here if needed
PROBE_WORKERS = 24       # parallel HTTP workers
PROBE_TIMEOUT_S = 8      # per-site connect+read timeout
MAX_BODY_BYTES = 300_000  # cap HTML read to avoid huge pages
LIVE_BODY_MIN = 1_200    # minimum byte count to consider a page "real"

def env(*keys):
    for k in keys:
        if os.environ.get(k): return os.environ[k]
    if SEC.exists():
        for ln in SEC.read_text().splitlines():
            ln = ln.strip()
            for k in keys:
                if ln.startswith(k + "="): return ln.split("=", 1)[1].strip().strip('"').strip("'")
    return ""

def get_json(url, hdr):
    return json.loads(urllib.request.urlopen(urllib.request.Request(url, headers=hdr), timeout=30).read())

# --- gather owned domains ---
domains = set()
# GA4 slugs
try:
    for r in json.loads((HERE / "cache" / "data.json").read_text())["rows"]:
        s = "".join(c for c in r["property"].lower().replace(" - ga4", "").replace("&", "and") if c.isalnum())
        if s: domains.add(s + ".com")
except Exception as e: print("ga4 slugs err:", e)
# Cloudflare
cft = env("CLOUDFLARE_API_TOKEN", "CF_API_TOKEN", "CF_TOKEN"); p = 1
while cft:
    r = get_json(f"https://api.cloudflare.com/client/v4/zones?per_page=50&page={p}", {"Authorization": "Bearer " + cft})
    domains |= {z["name"] for z in r.get("result", [])}
    if p >= r.get("result_info", {}).get("total_pages", 1): break
    p += 1
# GoDaddy
gk, gs = env("GODADDY_API_KEY", "GODADDY_KEY"), env("GODADDY_API_SECRET", "GODADDY_SECRET")
if gk and gs:
    try:
        domains |= {d["domain"] for d in get_json("https://api.godaddy.com/v1/domains?limit=1000&statuses=ACTIVE", {"Authorization": f"sso-key {gk}:{gs}"})}
    except Exception as e: print("godaddy err:", e)

# GSC granted sc-domains
gsc = set()
try:
    kw = json.loads((HERE / "cache" / "keywords.json").read_text())
    gsc = {s["url"].replace("sc-domain:", "") for s in kw.get("gsc_sites", []) if s["url"].startswith("sc-domain:")}
except Exception: pass

ctx = ssl.create_default_context(); ctx.check_hostname = False; ctx.verify_mode = ssl.CERT_NONE
def probe(d):
    row = {"domain": d, "http": 0, "live": False, "gtm": False, "ga4": False, "meta": False, "pin": False,
           "gsc": d in gsc}
    try:
        req = urllib.request.Request(f"https://{d}/", headers={"User-Agent": "Mozilla/5.0"})
        with urllib.request.urlopen(req, timeout=PROBE_TIMEOUT_S, context=ctx) as r:
            row["http"] = r.status
            final = r.geturl()
            body = r.read(MAX_BODY_BYTES).decode("utf-8", "ignore")
        # live = 200 + real body + not a cross-host redirect
        host = re.sub(r"^https?://(www\.)?", "", final).split("/")[0]
        row["live"] = row["http"] == 200 and len(body) > LIVE_BODY_MIN and d.replace("www.", "") in host
        row["gtm"] = bool(re.search(r"GTM-[A-Z0-9]{6,8}", body))
        row["ga4"] = bool(re.search(r"G-[A-Z0-9]{10}\b", body)) or "gtag(" in body
        row["meta"] = "fbq(" in body or "connect.facebook.net" in body
        row["pin"] = "pintrk(" in body or "s.pinimg.com/ct" in body
    except Exception as e:
        row["http"] = getattr(e, "code", 0)
    return row

doms = sorted(domains)
print(f"Probing {len(doms)} owned domains (parallel, $0 local)...\n")
with ThreadPoolExecutor(max_workers=PROBE_WORKERS) as ex:
    rows = list(ex.map(probe, doms))

live = [r for r in rows if r["live"]]
def yn(b): return "✅" if b else "—"
print(f"# Tracking coverage — {len(live)} LIVE of {len(rows)} owned domains\n")
print("| # | Domain | GSC | GA4 | GTM | Meta | Pinterest | missing |")
print("|---|--------|-----|-----|-----|------|-----------|---------|")
for i, r in enumerate(sorted(live, key=lambda x: (x["ga4"], x["meta"]), reverse=False), 1):
    miss = [k.upper() for k in ("ga4", "gtm", "meta", "pin") if not r[k]]
    print(f"| {i} | {r['domain']} | {yn(r['gsc'])} | {yn(r['ga4'])} | {yn(r['gtm'])} | {yn(r['meta'])} | {yn(r['pin'])} | {','.join(miss) or 'none'} |")

def pct(k): return f"{sum(1 for r in live if r[k])}/{len(live)}"
print(f"\n## Coverage (of {len(live)} live)")
print(f"- GSC: {pct('gsc')} · GA4: {pct('ga4')} · GTM: {pct('gtm')} · Meta: {pct('meta')} · Pinterest: {pct('pin')}")
zero = [r["domain"] for r in live if not (r["ga4"] or r["gtm"] or r["meta"] or r["pin"])]
print(f"\n## {len(zero)} LIVE sites with ZERO tracking (top priority):")
print("  " + ", ".join(zero))
# persist
out = HERE / "gtm" / "coverage-matrix.json"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(rows, indent=2))
print(f"\n(raw → {out})")