# Substack analytics → BigQuery: build guide

A complete, self-contained guide to pulling your **Substack podcast analytics**
into **Google BigQuery**, refreshed every 6 hours, with Slack alerts when it
breaks. Built by [Kickflip Analytics](https://kickflipanalytics.com).

> **How to use this file**
> Attach it to a **Claude Code** or **Claude Cowork** session, then paste the
> starter prompt from the [blog post that goes with it](https://kickflipanalytics.com/blog/substack-analytics-into-bigquery).
> Claude will walk you through the build in the Google Cloud console using your
> own details, explaining as it goes.
> You can also just follow it by hand — it's written to be read either way.

The result is two BigQuery tables, fully replaced on every run:

| Table | Contents |
|-------|----------|
| `podcast_downloads` | One row per date: `date`, `daily_downloads`, `cumulative_downloads`. |
| `podcast_episode_stats` | Full per-episode stats — opens, clicks, downloads, engagement, etc. |

```
Cloud Scheduler ──(every 6h)──► Cloud Function (Python) ──► BigQuery
                                    └── reads auth cookie from Secret Manager
Cloud Monitoring ──(on error)──► Slack
```

---

## 1. Prerequisites — get these before you start

- **A Google Cloud project with billing enabled.** The whole pipeline runs in
  the free tier / a few pennies a month, but billing must be switched on.
- **A Substack publication with a podcast**, and you logged into it in your
  browser (you need admin access to the stats).
- **Your session cookie** and **API details** — how to grab both is in section 2.
- **(Optional) A Slack workspace** if you want failure alerts.
- **(Optional) Claude Code or Claude Cowork** if you want the build done *with*
  you rather than by hand.

You do **not** need to install anything locally — every step is in the browser.

---

## 2. Find your API details and auth cookie

Substack doesn't publish this API, but the stats page in your browser uses it, so
you can read everything you need straight off the network tab.

> Credit where it's due: the trick of using the `connect.sid` session cookie for
> auth came from **Paweł** — see his note here:
> https://substack.com/@huryn/note/c-181571328

1. Log into your Substack publication and open your **podcast stats** page.
2. Open your browser **Developer Tools** (F12) → **Network** tab → **refresh**.
3. In the request list, find calls to
   **`/api/v1/publication/stats/podcast…`**. You'll see two you care about:
   - **Downloads:**
     `https://YOUR_PUBLICATION/api/v1/publication/stats/podcast/downloads?from=DATE&section_id=SECTION_ID`
   - **Per-episode:**
     `https://YOUR_PUBLICATION/api/v1/publication/stats/podcast?section_id=SECTION_ID&offset=0&limit=25&order_by=post_date&order_direction=desc`
4. From those requests, note down:
   - **`YOUR_PUBLICATION`** — your publication's host (e.g.
     `yourshow.substack.com`, or your custom domain).
   - **`SECTION_ID`** — the `section_id` query parameter.
5. Click one of those requests → **Headers** → **Request Headers** → find the
   **`Cookie:`** line → copy the **`connect.sid` value** (everything after
   `connect.sid=` up to the next `;`). **This is a secret** — treat it like a
   password. It goes into Secret Manager later, never into code.

**Tip:** because your browser is already logged in, you can paste either full URL
into the address bar to see the raw JSON — a quick way to confirm things work.

---

## 3. The code — two files

You'll paste these into the Cloud Function later. Nothing here is
publication-specific: every detail comes from environment variables you set at
deploy time, so the same code works for any Substack podcast.

### `main.py`

```python
"""
Substack podcast analytics -> BigQuery refresh.

Runs as a Cloud Function (2nd gen), triggered by Cloud Scheduler every 6 hours.
Each run FULLY REPLACES (WRITE_TRUNCATE) two BigQuery tables:
  * podcast_downloads       daily + cumulative downloads, one row per date
  * podcast_episode_stats   the full per-episode stats (paginated source)

The auth cookie is read from Secret Manager at runtime -- never hard-coded.
Alerting is handled by Cloud Monitoring: problems are written to the error log
(stderr => severity=ERROR) and a log-based alert forwards them to Slack.
"""

import datetime
import json
import os
import sys
import urllib.parse

import functions_framework
import requests
from google.cloud import bigquery, secretmanager

# ---- Configuration (set via environment variables at deploy time) ----
PROJECT_ID = os.environ["GCP_PROJECT_ID"]
PUBLICATION_URL = os.environ["PUBLICATION_URL"].rstrip("/")  # e.g. https://yourshow.substack.com
SECTION_ID = os.environ["SECTION_ID"]
DOWNLOADS_FROM = os.environ["DOWNLOADS_FROM"]  # ISO date at/before your podcast launch
BQ_DATASET = os.environ.get("BQ_DATASET", "substack_analytics")
BQ_LOCATION = os.environ.get("BQ_LOCATION", "US")
SECRET_ID = os.environ.get("COOKIE_SECRET_ID", "substack-cookie")

BASE = f"{PUBLICATION_URL}/api/v1/publication/stats/podcast"
PAGE_LIMIT = 25  # episodes per page; auto-shrinks if the API rejects it (400)


def log_error(message: str) -> None:
    """Write to stderr so Cloud Logging tags it severity=ERROR (drives alerts)."""
    print(f"PIPELINE ALERT: {message}", file=sys.stderr)


def get_cookie() -> str:
    """Read the connect.sid cookie value from Secret Manager."""
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{PROJECT_ID}/secrets/{SECRET_ID}/versions/latest"
    resp = client.access_secret_version(request={"name": name})
    return resp.payload.data.decode("utf-8").strip()


def api_get(url: str, cookie: str):
    """GET a URL with the auth cookie attached; return parsed JSON."""
    headers = {
        "Cookie": f"connect.sid={cookie}",
        "Accept": "application/json",
        "User-Agent": "substack-bigquery-pipeline/1.0",
    }
    r = requests.get(url, headers=headers, timeout=60)
    r.raise_for_status()
    return r.json()


def load_to_bq(rows: list, table: str, ingested_at: str) -> int:
    """Full-replace a BigQuery table from a list of JSON row dicts."""
    if not rows:
        log_error(f"0 rows for {table}; skipping load so the table isn't wiped.")
        return 0

    client = bigquery.Client(project=PROJECT_ID)
    table_id = f"{PROJECT_ID}.{BQ_DATASET}.{table}"

    for row in rows:
        row["_ingested_at"] = ingested_at

    job_config = bigquery.LoadJobConfig(
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
        source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
        autodetect=True,
    )
    job = client.load_table_from_json(
        rows, table_id, job_config=job_config, location=BQ_LOCATION
    )
    job.result()  # wait for completion; raises on failure
    return len(rows)


def fetch_downloads(cookie: str) -> list:
    """
    /downloads returns {daily:[[date,n]], cumulative:[[date,n]], episodes:[...]}.
    Merge daily + cumulative into one row per date; ISO-format the dates.
    (The thin `episodes` array is ignored -- the episodes endpoint has the full data.)
    """
    query = urllib.parse.urlencode({"from": DOWNLOADS_FROM, "section_id": SECTION_ID})
    payload = api_get(f"{BASE}/downloads?{query}", cookie)

    daily = {d: n for d, n in payload.get("daily", [])}
    cumulative = {d: n for d, n in payload.get("cumulative", [])}

    rows = []
    for date in sorted(set(daily) | set(cumulative)):
        rows.append(
            {
                "date": date.replace("/", "-"),  # 2026/07/14 -> 2026-07-14 (DATE)
                "daily_downloads": daily.get(date),
                "cumulative_downloads": cumulative.get(date),
            }
        )
    return rows


def fetch_episodes(cookie: str) -> list:
    """
    /podcast returns {rows:[{...}], total:N}. Page through all episodes and
    replace the "-" placeholders with null so numeric columns type correctly.
    """
    all_rows = []
    offset = 0
    limit = PAGE_LIMIT
    while True:
        query = urllib.parse.urlencode(
            {
                "section_id": SECTION_ID,
                "offset": offset,
                "limit": limit,
                "order_by": "post_date",
                "order_direction": "desc",
            }
        )
        try:
            payload = api_get(f"{BASE}?{query}", cookie)
        except requests.HTTPError as err:
            # The API caps page size; if it rejects this limit, halve and retry.
            status = err.response.status_code if err.response is not None else None
            if status == 400 and limit > 1:
                limit = max(1, limit // 2)
                continue
            raise

        rows = payload.get("rows", [])
        if not rows:
            break
        all_rows.extend(rows)
        if len(rows) < limit:
            break
        offset += limit

    for row in all_rows:
        for key, value in list(row.items()):
            if value == "-":
                row[key] = None  # placeholder for "no data"
    return all_rows


@functions_framework.http
def refresh(request):
    """HTTP entry point. Cloud Scheduler calls this every 6 hours."""
    ingested_at = datetime.datetime.now(datetime.timezone.utc).isoformat()

    try:
        cookie = get_cookie()
        n_downloads = load_to_bq(
            fetch_downloads(cookie), "podcast_downloads", ingested_at
        )
        n_episodes = load_to_bq(
            fetch_episodes(cookie), "podcast_episode_stats", ingested_at
        )
    except requests.HTTPError as exc:
        code = exc.response.status_code if exc.response is not None else "?"
        if code in (401, 403):
            log_error(
                f"auth failed ({code}) -- the cookie has likely expired. "
                f"Add a new version to the '{SECRET_ID}' secret."
            )
        else:
            log_error(f"API request failed ({code}): {exc}")
        raise  # re-raise so the run is marked failed and logged at ERROR
    except Exception as exc:
        log_error(f"pipeline failed: {type(exc).__name__}: {exc}")
        raise

    result = {
        "downloads_rows": n_downloads,
        "episode_rows": n_episodes,
        "ingested_at": ingested_at,
    }
    print(json.dumps(result))
    return (json.dumps(result), 200, {"Content-Type": "application/json"})
```

### `requirements.txt`

```
functions-framework>=3.5.0
google-cloud-bigquery>=3.25.0
google-cloud-secret-manager>=2.20.0
requests>=2.32.0
```

---

## 4. Build it in the Google Cloud console

Everything below is point-and-click at https://console.cloud.google.com. Make
sure the **project selector** at the top shows your project. Pick a **region**
and use it everywhere — `US` (BigQuery multi-region) with `us-central1`, or
`europe-west2` (London) for the UK, etc. Replace the `YOUR_…` placeholders.

### Phase A — enable services & create the dataset
1. **APIs & Services → Enabled APIs & services → + Enable APIs and Services.**
   Enable each: **BigQuery API, Cloud Functions API, Cloud Run Admin API,
   Cloud Build API, Cloud Scheduler API, Secret Manager API.**
2. **BigQuery** → in the Explorer, click **⋮** next to your project →
   **Create dataset.** Dataset ID: `substack_analytics`. Location: your chosen
   region. **Create dataset.** (The tables get created automatically by the code.)

### Phase B — store the cookie
1. **Secret Manager → + Create Secret.** Name: `substack-cookie`. Value: the
   `connect.sid` value from section 2. **Create Secret.**

### Phase C — create the Cloud Function
1. **Cloud Run functions → Create function.** Environment: 2nd gen.
   Name: `substack-refresh`. Region: your chosen region.
2. **Trigger:** the HTTPS endpoint is automatic — **do not add a trigger.**
   Set authentication to **Require authentication** (do not allow unauthenticated).
3. Expand **Runtime, build, connections and security settings**:
   - **Memory** 512 MiB, **Timeout** 540 s.
   - **Runtime environment variables** — add:

     | Name | Value |
     |------|-------|
     | `GCP_PROJECT_ID` | `YOUR_PROJECT_ID` |
     | `PUBLICATION_URL` | `https://YOUR_PUBLICATION` |
     | `SECTION_ID` | `YOUR_SECTION_ID` |
     | `DOWNLOADS_FROM` | `YOUR_LAUNCH_DATE` e.g. `2026-01-01T00:00:00.000Z` |
     | `BQ_DATASET` | `substack_analytics` |
     | `BQ_LOCATION` | `YOUR_REGION` e.g. `US` or `europe-west2` |

   - **Note the runtime service account** shown (usually the Compute Engine
     default service account, `NNN-compute@developer.gserviceaccount.com`) —
     you'll grant it access next.
4. **Next.** Runtime **Python 3.12**, **Entry point** `refresh`. In the inline
   editor, replace `main.py` and `requirements.txt` with the code from section 3.
   **Deploy** (takes a few minutes).
5. **Let it read the secret:** Secret Manager → `substack-cookie` →
   **Permissions → Grant Access** → principal: the runtime service account →
   role: **Secret Manager Secret Accessor** → Save.
6. **Let it write to BigQuery:** IAM & Admin → IAM → Grant Access (same service
   account) → roles: **BigQuery Data Editor** and **BigQuery Job User** → Save.
7. **Test:** the function's **Testing** tab → **Test the function** (empty body).
   Expect `{"downloads_rows": N, "episode_rows": N, ...}`. Then check the tables
   in BigQuery → **Preview**. (Permission grants can take a minute to propagate —
   if the first test errors, wait and retry.)

### Phase D — schedule it every 6 hours
1. **Cloud Scheduler → Create job.** Name: `substack-refresh-6h`. Region: yours.
   Frequency: `0 */6 * * *`. Timezone: your choice. **Continue.**
2. **Target type** HTTP. **URL:** the function's URL (its **Trigger** tab).
   **Method** GET. **Auth header:** **Add OIDC token.** **Service account:** the
   runtime service account. **Audience:** the same function URL. **Create.**
3. **Let the scheduler invoke it:** Cloud Run → `substack-refresh` service →
   **Permissions** → Add principal: the runtime service account →
   role: **Cloud Run Invoker** → Save.
4. **Test now:** Cloud Scheduler → the job → **Force run** → re-check the tables.

### Phase E — Slack alerts (recommended)
1. **Monitoring → Alerting → Edit notification channels → Slack → Add new.**
   Authorise the *Google Cloud Monitoring* Slack app, pick a channel (for private
   channels, also `/invite @Google Cloud Monitoring`), give it a display name.
2. **Logs Explorer** → run this query:
   ```
   resource.type="cloud_run_revision"
   resource.labels.service_name="substack-refresh"
   severity>=ERROR
   ```
   → **Create alert** → name it `Substack pipeline errors` → notification channel:
   your Slack channel → Save.
3. **Test:** use **Send test notification** on the channel. (Optional end-to-end:
   add a junk cookie version, Force-run the job to trigger a 401, then restore the
   real cookie.)

---

## 5. Maintenance

- **The cookie will expire.** When alerts show a `401`, get a fresh `connect.sid`
  (section 2) → Secret Manager → `substack-cookie` → **+ New Version** → paste →
  Add. The function reads the latest version next run — no redeploy needed.
- **Logs:** the function's **Logs** tab.
- **Change cadence:** edit the Cloud Scheduler job's frequency.
- **Full-replace safety:** every run truncates and reloads. If an endpoint
  returns 0 rows, the load is skipped so a good table isn't wiped by a transient
  error — and you get an alert.

---

*This is an unofficial, undocumented API. Substack can change it at any time and
this may stop working. Use it for your own data, sensibly.*