#!/usr/bin/env python3 """ Reproduction script for the CFG AI Search Impressions Study (June 12 to July 11, 2026). Input : seo/gsc-query-page-20260612-20260711.csv (3,020 rows, GSC API export, query + page + clicks + impressions + ctr + position) Output: data-downloads/ai-search-impressions-study-2026-07.csv (aggregate table) plus a printed report of every number cited on /resources/ai-search-impressions-study-2026/. Classification regex source: seo/measurement.md (Phase 3 bot-exclusion filter, established 2026-07-12). Every query row is assigned to exactly ONE category, first match wins, in the precedence order below. Run from repo root: python3 data-downloads/build-ai-search-impressions-study.py No dependencies beyond the Python 3 standard library. """ import csv import re import sys from collections import defaultdict from pathlib import Path ROOT = Path(__file__).resolve().parent.parent SRC = ROOT / "seo" / "gsc-query-page-20260612-20260711.csv" OUT = ROOT / "data-downloads" / "ai-search-impressions-study-2026-07.csv" # Category definitions, first match wins (order matters). CATEGORIES = [ ("prompt_injection_fragments", re.compile(r"context:\s*location|do not include location", re.I)), ("search_operator_strings", re.compile(r"site:|\\n", re.I)), ("bracketed_llm_queries", re.compile(r"^\[.*\]$")), ("boolean_chains", re.compile(r'"\s+or\s+"|\)\s+\(', re.I)), ("quoted_strings", re.compile(r'^".*"$')), ("spaced_out_letters", re.compile(r"^c(\s\w)+", re.I)), ("machine_cluster_belize", re.compile(r"belize\s.*(2026|benchmark|hipaa|phipa)", re.I)), ("conversational_fragments", re.compile( r"^(yes|no|yeah|oui|ok|okay|all|da|sure|si)$" r"|^(compare|list|evaluate|which|what operational|where can i|when does|is it (what|how))\b.*[.?]$" r"|^our (coo|ceo|cfo)\b" r"|^can someone guide" r"|^for each of the following" r"|^how (do|does)\b.*\?$", re.I)), ] def classify(query: str) -> str: for name, pattern in CATEGORIES: if pattern.search(query): return name return "human" def main() -> None: with open(SRC, newline="", encoding="utf-8") as f: rows = list(csv.DictReader(f)) assert len(rows) == 3020, f"expected 3,020 data rows, got {len(rows)}" for r in rows: r["clicks"] = int(r["clicks"].replace(",", "")) r["impressions"] = int(r["impressions"].replace(",", "")) r["position"] = float(r["position"]) r["category"] = classify(r["query"]) total_impr = sum(r["impressions"] for r in rows) total_clicks = sum(r["clicks"] for r in rows) agg = defaultdict(lambda: { "rows": 0, "queries": set(), "clicks": 0, "impressions": 0, "pos_impr_sum": 0.0, "impr_pos_top3": 0, "impr_pos_4_10": 0, "impr_pos_11_20": 0, "impr_pos_21plus": 0, }) for r in rows: a = agg[r["category"]] a["rows"] += 1 a["queries"].add(r["query"]) a["clicks"] += r["clicks"] a["impressions"] += r["impressions"] a["pos_impr_sum"] += r["position"] * r["impressions"] p, i = r["position"], r["impressions"] if p <= 3: a["impr_pos_top3"] += i elif p <= 10: a["impr_pos_4_10"] += i elif p <= 20: a["impr_pos_11_20"] += i else: a["impr_pos_21plus"] += i order = [name for name, _ in CATEGORIES] + ["human"] header = ["category", "rows", "unique_queries", "clicks", "impressions", "impression_share_pct", "ctr_pct", "impr_weighted_position", "impr_share_pos_1_3_pct", "impr_share_pos_4_10_pct", "impr_share_pos_11_20_pct", "impr_share_pos_21plus_pct"] out_rows = [] for cat in order: a = agg[cat] impr = a["impressions"] out_rows.append([ cat, a["rows"], len(a["queries"]), a["clicks"], impr, round(100 * impr / total_impr, 2), round(100 * a["clicks"] / impr, 2) if impr else 0.0, round(a["pos_impr_sum"] / impr, 1) if impr else 0.0, round(100 * a["impr_pos_top3"] / impr, 1) if impr else 0.0, round(100 * a["impr_pos_4_10"] / impr, 1) if impr else 0.0, round(100 * a["impr_pos_11_20"] / impr, 1) if impr else 0.0, round(100 * a["impr_pos_21plus"] / impr, 1) if impr else 0.0, ]) nh_cats = [c for c in order if c != "human"] nh_impr = sum(agg[c]["impressions"] for c in nh_cats) nh_clicks = sum(agg[c]["clicks"] for c in nh_cats) nh_rows = sum(agg[c]["rows"] for c in nh_cats) nh_pos = sum(agg[c]["pos_impr_sum"] for c in nh_cats) / nh_impr out_rows.append([ "ALL_NON_HUMAN", nh_rows, sum(len(agg[c]["queries"]) for c in nh_cats), nh_clicks, nh_impr, round(100 * nh_impr / total_impr, 2), round(100 * nh_clicks / nh_impr, 2), round(nh_pos, 1), round(100 * sum(agg[c]["impr_pos_top3"] for c in nh_cats) / nh_impr, 1), round(100 * sum(agg[c]["impr_pos_4_10"] for c in nh_cats) / nh_impr, 1), round(100 * sum(agg[c]["impr_pos_11_20"] for c in nh_cats) / nh_impr, 1), round(100 * sum(agg[c]["impr_pos_21plus"] for c in nh_cats) / nh_impr, 1), ]) out_rows.append([ "TOTAL", len(rows), len({r["query"] for r in rows}), total_clicks, total_impr, 100.0, round(100 * total_clicks / total_impr, 2), round(sum(r["position"] * r["impressions"] for r in rows) / total_impr, 1), "", "", "", "", ]) OUT.parent.mkdir(exist_ok=True) with open(OUT, "w", newline="", encoding="utf-8") as f: w = csv.writer(f) w.writerow(header) w.writerows(out_rows) # ---- printed report ---- wcol = max(len(c) for c in order) + 2 print(f"Source rows: {len(rows)} clicks {total_clicks} impressions {total_impr}") print(f"{'category':<{wcol}}{'rows':>6}{'uniq':>6}{'clicks':>8}{'impr':>8}" f"{'impr%':>8}{'ctr%':>8}{'w.pos':>7}{'<=3%':>7}{'4-10%':>7}") for row in out_rows: print(f"{row[0]:<{wcol}}{row[1]:>6}{row[2]:>6}{row[3]:>8}{row[4]:>8}" f"{row[5]:>8}{row[6]:>8}{row[7]:>7}{str(row[8]):>7}{str(row[9]):>7}") # Non-human queries at strong positions strong = [r for r in rows if r["category"] != "human" and r["position"] <= 10] print(f"\nNon-human rows at position <= 10: {len(strong)}, " f"impressions {sum(r['impressions'] for r in strong)}, " f"clicks {sum(r['clicks'] for r in strong)}") # Top example queries per non-human category (aggregated across pages) print("\n=== Example queries (top by impressions, per category) ===") by_q = defaultdict(lambda: [0, 0, 0.0, ""]) # impr, clicks, best pos, cat for r in rows: if r["category"] == "human": continue e = by_q[r["query"]] e[0] += r["impressions"] e[1] += r["clicks"] e[2] = min(e[2], r["position"]) if e[2] else r["position"] e[3] = r["category"] ranked = sorted(by_q.items(), key=lambda kv: -kv[1][0]) per_cat = defaultdict(int) shown = 0 for q, (impr, clicks, pos, cat) in ranked: if per_cat[cat] >= 4 or shown >= 24: continue per_cat[cat] += 1 shown += 1 print(f"[{cat}] impr={impr} clicks={clicks} best_pos={pos:.1f}\n {q!r}") print(f"\nWrote {OUT.relative_to(ROOT)}") if __name__ == "__main__": main()