← back to Ga Allsites
scripts/ga4_absence_detect.py
199 lines
#!/usr/bin/env python3
"""
Hardened GA4 absence detector — the trustworthy feeder for create_inject_properties.py.
Replaces the naive _verify_ga4_gap.py (live-HTML-only) which produced two classes of
false positives that led to wasted/duplicate/orphan GA4 properties (2026-08-18):
1. DUPLICATE properties — a site already had GA4 wired SERVER-SIDE (EJS `GA_ID`,
a committed gtag in the repo) that a live-HTML probe missed (not deployed yet,
or Cloudflare/parking cache masked it). We minted a second property for it.
2. ORPHAN properties — a PARKED / for-sale / dead domain (registrar AWS lander)
has no serving layer of ours, so a minted property tracks nothing.
This detector classifies every candidate into buckets and emits ONLY the genuinely
injectable ones (LIVE, no GA4 in live HTML, no GA4 already wired in the local repo)
to cache/_ga4_inject_targets.json — so the mint step can't manufacture dups/orphans.
$0 local: HTTP GETs + local repo greps only. No paid API, no LLM.
Usage:
python3 scripts/ga4_absence_detect.py # candidates = current targets file (verify mode)
python3 scripts/ga4_absence_detect.py --portfolio # candidates = cache/_portfolio_domains.json ["union"]
python3 scripts/ga4_absence_detect.py --file PATH # candidates = JSON list at PATH
python3 scripts/ga4_absence_detect.py --write # actually overwrite the targets file (default: dry-run)
"""
from __future__ import annotations
import json
import pathlib
import re
import ssl
import subprocess
import sys
import urllib.request
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
ROOT = pathlib.Path(__file__).resolve().parent.parent
CACHE = ROOT / "cache"
PROJECTS = pathlib.Path.home() / "Projects"
CTX = ssl.create_default_context()
CTX.check_hostname = False
CTX.verify_mode = ssl.CERT_NONE
# Broad GA4/GTM/UA/dataLayer detector for LIVE HTML (same coverage as _verify_ga4_gap's Cody-fix).
# Note: a real GA4 measurement id is G- + EXACTLY 10 alphanumerics; the loose {6,} is fine here
# because the other alternatives (gtag/gtm/dataLayer) confirm a real tag is present.
GA4 = re.compile(
r"(G-[A-Z0-9]{10}\b|googletagmanager\.com/gtag/js|googletagmanager\.com/gtm\.js"
r"|gtag\('config'|GTM-[A-Z0-9]+|UA-\d{4,}|google-analytics\.com/analytics\.js|dataLayer)"
)
# Strict GA4 measurement-id shape for the LOCAL-repo grep (avoids coincidental short/hex matches
# like a bogus "G-51302138"): G- followed by EXACTLY 10 uppercase-alnum, on a word boundary.
GA4_ID = re.compile(r"G-[A-Z0-9]{10}\b")
# Parked / for-sale / placeholder / default-server pages — NOT injectable (would orphan a property).
PARK = re.compile(
r"(domain (is )?for sale|buy this domain|parked|godaddy\.com/domainsearch|afternic"
r"|sedoparking|this domain may be for sale|hugedomains|dan\.com|is coming soon"
r"|website coming soon|under construction|default web page|welcome to nginx"
r"|apache2 ubuntu default|namecheap|parkingcrew|bodis)",
re.IGNORECASE,
)
# Known registrar/parking host IP prefixes (AWS landers seen on the parked tail).
PARK_IP_PREFIXES = ("3.33.", "15.197.", "3.64.", "13.248.", "76.223.")
def dslug(d: str) -> str:
return "".join(c for c in d.rsplit(".", 1)[0].lower() if c.isalnum())
def repo_dirs(domain: str) -> list:
"""Candidate local repo dirs for a domain, WITHOUT colliding stem across TLDs.
Always try the exact full-domain dir (~/Projects/<domain>). Only fall back to the
bare stem (~/Projects/<stem>) for .com domains, where stem->repo is the established
convention — otherwise 'designerwallcoverings.ai' would wrongly match the
'designerwallcoverings' (.com) repo.
"""
stem, _, tld = domain.rpartition(".")
cands = [PROJECTS / domain]
if tld == "com":
cands.append(PROJECTS / stem)
return [c for c in cands if c.is_dir()]
def repo_ga4(domain: str) -> str | None:
"""Return an existing GA4 id if the domain's LOCAL repo already wires gtag/GA_ID.
Catches the duplicate-property class: GA4 committed in served templates/server
(EJS GA_ID, static index.html) even if the live probe missed it. Scans only the
SERVED files (public/, views/, root server.js + root *.html) — never node_modules —
and accepts only a strict 10-char GA4 measurement id.
"""
for cand in repo_dirs(domain):
files: list = []
for sub in ("public", "views"):
d = cand / sub
if d.is_dir():
for ext in ("*.html", "*.ejs", "*.pug"):
files += [str(p) for p in d.rglob(ext)]
files += [str(p) for p in cand.glob("*.html")]
if (cand / "server.js").is_file():
files.append(str(cand / "server.js"))
for f in files:
try:
m = GA4_ID.search(pathlib.Path(f).read_text(errors="ignore"))
except Exception:
continue
if m:
return m.group(0)
return None
def host_ip(domain: str) -> str:
try:
return subprocess.run(["dig", "+short", domain, "A"],
capture_output=True, text=True, timeout=8).stdout.strip().splitlines()[-1]
except Exception:
return ""
def probe(domain: str) -> tuple:
"""Return (domain, bucket, existing_ga4_or_ip)."""
# Local repo is authoritative for 'already wired' regardless of live state.
rid = repo_ga4(domain)
html = None
final = ""
for scheme in ("https", "http"):
try:
req = urllib.request.Request(f"{scheme}://{domain}/",
headers={"User-Agent": "Mozilla/5.0 (ga4-detect)"})
r = urllib.request.urlopen(req, timeout=12, context=CTX)
final = r.geturl()
html = r.read(250000).decode("utf-8", "ignore")
break
except Exception:
continue
if rid:
return (domain, "ALREADY_HAS_GA4_LOCAL", rid)
if html is None:
ip = host_ip(domain)
return (domain, "UNREACHABLE", ip)
if GA4.search(html):
m = re.search(r"G-[A-Z0-9]{6,}", html)
return (domain, "ALREADY_HAS_GA4_LIVE", m.group(0) if m else "yes")
# No GA4 in html and none in repo — is it a real site or a parked lander?
off = dslug(domain) not in dslug(final.split("//")[-1].split("/")[0]) if final else False
ip = host_ip(domain)
if PARK.search(html) or len(html) < 400 or any(ip.startswith(p) for p in PARK_IP_PREFIXES):
return (domain, "PARKED", ip)
if off:
return (domain, "REDIRECT", final)
return (domain, "NEEDS_INJECT", "")
def main() -> None:
args = sys.argv[1:]
write = "--write" in args
if "--portfolio" in args:
cands = json.loads((CACHE / "_portfolio_domains.json").read_text())["union"]
src = "portfolio union"
elif "--file" in args:
p = args[args.index("--file") + 1]
cands = json.loads(pathlib.Path(p).read_text())
src = p
else:
cands = json.loads((CACHE / "_ga4_inject_targets.json").read_text())
src = "current _ga4_inject_targets.json (verify mode)"
res = list(ThreadPoolExecutor(max_workers=20).map(probe, cands))
buckets: dict[str, list] = defaultdict(list)
for d, cls, info in res:
buckets[cls].append((d, info))
order = ["NEEDS_INJECT", "ALREADY_HAS_GA4_LOCAL", "ALREADY_HAS_GA4_LIVE",
"PARKED", "REDIRECT", "UNREACHABLE"]
print(f"GA4 absence detection — candidates: {len(cands)} from {src}\n")
for k in order:
rows = sorted(buckets.get(k, []))
print(f" {k:24} {len(rows)}")
for d, info in rows:
print(f" {d:38} {info}")
needs = sorted(d for d, _ in buckets.get("NEEDS_INJECT", []))
# Durable buckets for the mint step + reconciliation.
(CACHE / "_ga4_detect_buckets.json").write_text(json.dumps(
{k: sorted(buckets.get(k, [])) for k in order}, indent=2))
print(f"\nTRUE inject targets (clean): {len(needs)}")
if write:
(CACHE / "_ga4_inject_targets.json").write_text(json.dumps(needs))
print(f"WROTE cache/_ga4_inject_targets.json ({len(needs)} domains)")
else:
print("(dry-run — pass --write to overwrite cache/_ga4_inject_targets.json)")
if __name__ == "__main__":
main()