Google Search Console skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "Google Search Console"
  short_description: "Explain the drop, then find the next clicks"
  default_prompt: "$google-search-console Here are this month's and last month's Queries.csv, why did clicks drop?"

#!/usr/bin/env python3
"""Analyze Google Search Console performance exports. Standard library only.

Accepts the UI export CSVs (Queries.csv / Pages.csv from the Performance report zip) and
API/BigQuery-style CSVs with columns such as query, page, clicks, impressions, ctr, position.

  python3 scripts/gsc_analyze.py Queries.csv [--brand 'acme|acme ?app'] [--min-impr 100]
  python3 scripts/gsc_analyze.py Queries.csv --compare Queries_prev.csv
  python3 scripts/gsc_analyze.py query_page.csv            # query+page rows: cannibalization

Sections: totals (branded vs non-branded), striking-distance rows (position 8-20),
CTR outliers against the site's own CTR-by-position curve, period-over-period movers,
and queries split across several pages.
Exit codes: 0 ok, 2 unreadable input or missing columns.
"""
from __future__ import annotations

import argparse
import csv
import re
import statistics
import sys
from collections import defaultdict

ALIASES = {
    "query": ["top queries", "query", "queries", "search query"],
    "page": ["top pages", "page", "pages", "url", "landing page"],
    "clicks": ["clicks", "url_clicks"],
    "impressions": ["impressions"],
    "ctr": ["ctr"],
    "position": ["position", "avg. position", "average position", "sum_position"],
}


def num(value: str) -> float:
    v = (value or "").strip().replace(",", "")
    if v.endswith("%"):
        return float(v[:-1]) / 100
    return float(v) if v else 0.0


def load(path: str) -> tuple[list[dict], str]:
    with open(path, encoding="utf-8-sig", newline="") as fh:
        reader = csv.DictReader(fh)
        if not reader.fieldnames:
            raise ValueError("empty file")
        mapping = {}
        lower = {f.lower().strip(): f for f in reader.fieldnames}
        for key, names in ALIASES.items():
            for n in names:
                if n in lower:
                    mapping[key] = lower[n]
                    break
        missing = [k for k in ("clicks", "impressions") if k not in mapping]
        if missing or not ({"query", "page"} & mapping.keys()):
            raise ValueError(f"cannot find columns {missing or ['query/page']} in {reader.fieldnames}; "
                             "rename headers to English (query/page, clicks, impressions, ctr, position)")
        rows = []
        for r in reader:
            row = {k: r[v] for k, v in mapping.items()}
            row["clicks"] = num(row["clicks"])
            row["impressions"] = num(row["impressions"])
            row["position"] = num(row.get("position", "0"))
            if mapping.get("position") and "sum_position" in mapping["position"].lower() and row["impressions"]:
                row["position"] = row["position"] / row["impressions"] + 1  # BigQuery sum_position is zero-based
            row["ctr"] = row["clicks"] / row["impressions"] if row["impressions"] else 0.0
            rows.append(row)
    key = "query" if "query" in mapping else "page"
    return rows, key


def fmt_row(r: dict, key: str) -> str:
    return (f"| {r[key][:70]} | {int(r['clicks'])} | {int(r['impressions'])} | "
            f"{r['ctr']:.1%} | {r['position']:.1f} |")


def table(rows: list[dict], key: str, limit: int) -> None:
    print(f"| {key} | clicks | impressions | ctr | position |\n|---|---|---|---|---|")
    for r in rows[:limit]:
        print(fmt_row(r, key))
    if not rows:
        print("| (none) | | | | |")
    print()


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("csv")
    ap.add_argument("--compare", help="same export for the previous period")
    ap.add_argument("--brand", help="regex for branded queries, case-insensitive")
    ap.add_argument("--min-impr", type=int, default=100, help="minimum impressions for opportunity lists")
    ap.add_argument("--limit", type=int, default=25)
    args = ap.parse_args()
    try:
        rows, key = load(args.csv)
    except (OSError, ValueError) as exc:
        print(f"error: {args.csv}: {exc}", file=sys.stderr)
        return 2

    brand = re.compile(args.brand, re.I) if args.brand else None
    print(f"# Search Console analysis: {args.csv}\n")

    tot_c = sum(r["clicks"] for r in rows)
    tot_i = sum(r["impressions"] for r in rows)
    print(f"Rows: {len(rows)} ({key} level) · clicks {int(tot_c):,} · impressions {int(tot_i):,} "
          f"· CTR {tot_c / tot_i if tot_i else 0:.2%}")
    print("Note: exported rows omit anonymized queries, so row totals are below the report total.\n")
    if brand and key == "query":
        b = [r for r in rows if brand.search(r["query"])]
        bc, bi = sum(r["clicks"] for r in b), sum(r["impressions"] for r in b)
        print(f"Branded: {len(b)} queries, {int(bc):,} clicks ({bc / tot_c if tot_c else 0:.0%} of clicks), "
              f"{int(bi):,} impressions. Non-branded analysis below excludes them.\n")
        rows = [r for r in rows if not brand.search(r["query"])]

    has_pos = any(r["position"] for r in rows)
    if has_pos:
        print("## Striking distance (position 8-20, by impressions)\n")
        sd = sorted((r for r in rows if 8 <= r["position"] <= 20 and r["impressions"] >= args.min_impr),
                    key=lambda r: -r["impressions"])
        table(sd, key, args.limit)

        print("## CTR below this site's norm for the position\n")
        buckets = defaultdict(list)
        for r in rows:
            if r["impressions"] >= args.min_impr and r["position"] >= 1:
                buckets[min(int(round(r["position"])), 20)].append(r["ctr"])
        norms = {b: statistics.median(v) for b, v in buckets.items() if len(v) >= 5}
        print("Median CTR by rounded position (rows with enough impressions): " +
              ", ".join(f"{b}: {norms[b]:.1%}" for b in sorted(norms)) + "\n")
        low = []
        for r in rows:
            b = min(int(round(r["position"])), 20)
            if r["impressions"] >= args.min_impr and b in norms and r["position"] <= 10 and r["ctr"] < 0.5 * norms[b]:
                r["missed"] = (norms[b] - r["ctr"]) * r["impressions"]
                low.append(r)
        low.sort(key=lambda r: -r["missed"])
        print("Sorted by estimated missed clicks. Check SERP features and intent before rewriting titles.\n")
        table(low, key, args.limit)

    if args.compare:
        try:
            prev, pkey = load(args.compare)
        except (OSError, ValueError) as exc:
            print(f"error: {args.compare}: {exc}", file=sys.stderr)
            return 2
        if brand and pkey == "query":
            prev = [r for r in prev if not brand.search(r["query"])]
        before = {r[pkey]: r for r in prev}
        now = {r[key]: r for r in rows}
        deltas = []
        for k in set(before) | set(now):
            c1, c0 = now.get(k, {}).get("clicks", 0), before.get(k, {}).get("clicks", 0)
            deltas.append((c1 - c0, k, before.get(k), now.get(k)))
        print("## Biggest click losses vs previous period\n")
        print("| key | clicks before | clicks now | Δ | position before → now |\n|---|---|---|---|---|")
        for d, k, b, n in sorted(deltas)[: args.limit]:
            if d >= 0:
                break
            pb = f"{b['position']:.1f}" if b else "-"
            pn = f"{n['position']:.1f}" if n else "gone"
            print(f"| {k[:70]} | {int(b['clicks']) if b else 0} | {int(n['clicks']) if n else 0} | {int(d)} | {pb} → {pn} |")
        print("\n## Biggest click gains\n")
        print("| key | clicks before | clicks now | Δ |\n|---|---|---|---|")
        for d, k, b, n in sorted(deltas, reverse=True)[: args.limit]:
            if d <= 0:
                break
            print(f"| {k[:70]} | {int(b['clicks']) if b else 0} | {int(n['clicks']) if n else 0} | +{int(d)} |")
        print()

    if rows and "query" in rows[0] and "page" in rows[0]:
        print("## Queries split across several pages (possible cannibalization)\n")
        by_q = defaultdict(list)
        for r in rows:
            by_q[r["query"]].append(r)
        found = []
        for q, rs in by_q.items():
            total = sum(r["impressions"] for r in rs)
            strong = [r for r in rs if total and r["impressions"] / total >= 0.2]
            if len(strong) >= 2 and total >= args.min_impr:
                found.append((total, q, strong))
        found.sort(reverse=True)
        for total, q, strong in found[: args.limit]:
            pages = "; ".join(f"{r['page']} (pos {r['position']:.1f}, {r['impressions'] / total:.0%})" for r in strong)
            print(f"- **{q}** ({int(total)} impr): {pages}")
        if not found:
            print("- none above thresholds")
        print()
    return 0


if __name__ == "__main__":
    sys.exit(main())