# Export campaign analytics to your data warehouse

A nightly script that pulls daily stats, sender breakdown, step funnel and analysis for every campaign into CSVs, with Postgres and BigQuery loaders.

Works with: Postgres, BigQuery, Metabase, cron.

Endpoints used:

- [`GET /v1/campaigns`](https://docs.versionseven.ai/api-reference/campaigns/list-campaigns) List campaigns
- [`GET /v1/campaigns/{campaign_id}/daily-stats`](https://docs.versionseven.ai/api-reference/campaigns/get-daily-stats) Get daily stats
- [`GET /v1/campaigns/{campaign_id}/senders/breakdown`](https://docs.versionseven.ai/api-reference/campaigns/get-sender-breakdown) Get the sender breakdown
- [`GET /v1/campaigns/{campaign_id}/step-funnel`](https://docs.versionseven.ai/api-reference/campaigns/get-step-funnel) Get the step funnel
- [`GET /v1/campaigns/{campaign_id}/analysis`](https://docs.versionseven.ai/api-reference/campaigns/get-campaign-analysis) Get campaign analytics

The app's Analytics tab answers questions about one campaign at a time. A warehouse answers the others: outbound next to pipeline and revenue, every campaign on one chart, a quarter compared with the last. This recipe is the export: a script that runs nightly, reads five analytics endpoints for every campaign, and writes one CSV per table, plus the SQL that loads them into Postgres or BigQuery without duplicating rows when a day is re-pulled.

## Before you start

- An API key with `campaigns:read` in `VICTORIA_API_KEY`. Nothing here writes to Victoria AI. See [Authentication](https://docs.versionseven.ai/guides/authentication).
- A warehouse: the loaders below are for Postgres (`psql`) and BigQuery (`bq`); any tool that loads CSV works.
- Node.js 18 or later, or Python 3.10 or later with `requests`.

## What it exports

| File | Source | Grain | Merge key |
| - | - | - | - |
| `daily_stats_by_sender.csv` | [`GET /v1/campaigns/{campaign_id}/daily-stats`](https://docs.versionseven.ai/api-reference/campaigns/get-daily-stats) with `group_by=sender` | campaign × day × sender | `campaign_id, day, sender_key` |
| `daily_stats_by_variation.csv` | the same with `group_by=variation` | campaign × day × A/B arm | `campaign_id, day, variation` |
| `sender_breakdown.csv` | [`GET /v1/campaigns/{campaign_id}/senders/breakdown`](https://docs.versionseven.ai/api-reference/campaigns/get-sender-breakdown) | campaign × sender, over the window | `exported_at, campaign_id, sender_key` |
| `step_funnel.csv` | [`GET /v1/campaigns/{campaign_id}/step-funnel`](https://docs.versionseven.ai/api-reference/campaigns/get-step-funnel) | campaign × step, over the window | `exported_at, campaign_id, step_id` |
| `campaign_analysis.csv` | [`GET /v1/campaigns/{campaign_id}/analysis`](https://docs.versionseven.ai/api-reference/campaigns/get-campaign-analysis) | campaign, over the window | `exported_at, campaign_id` |

The daily tables are the time series; the script re-pulls the trailing `DAYS` days (default 3) every night because the most recent day's row carries `partial: true` while the day is still running, and yesterday's can still change for a few hours. The merge key makes a re-pull an update, not a duplicate. The other three are snapshots of a window (default `30d`), stamped with `exported_at`, so trend charts can be built from the daily tables and the latest snapshot answers "where is it now".

Everything is per lead unless the column says messages, as in the app. [Analytics metrics](https://docs.versionseven.ai/help/analytics-metrics) defines each column.

## The script

**Node.js**

```javascript
// export-analytics.mjs
import { mkdir, writeFile } from "node:fs/promises";
import path from "node:path";

const API_URL = "https://api.versionseven.ai/v1";
const API_KEY = process.env.VICTORIA_API_KEY;
const DAYS = Number(process.env.DAYS ?? 3); // trailing days to re-pull for the daily tables
const WINDOW = process.env.WINDOW ?? "30d"; // date_filter for the snapshot tables
const OUT = process.env.OUT ?? "export";

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

async function call(path) {
  for (let attempt = 1; ; attempt++) {
    const response = await fetch(`${API_URL}${path}`, { headers: { Authorization: `Bearer ${API_KEY}` } });
    const data = await response.json().catch(() => ({}));
    if ((response.status !== 429 && response.status !== 503) || attempt === 5) {
      if (response.status !== 200) throw new Error(`${path} answered ${response.status} ${data.error ?? ""}`);
      return data;
    }
    await sleep((Number(response.headers.get("Retry-After")) || 2 ** attempt) * 1000);
  }
}

async function listCampaigns() {
  const campaigns = [];
  for (let offset = 0; ; ) {
    const data = await call(`/campaigns?limit=500&offset=${offset}`);
    campaigns.push(...data.campaigns);
    if (!data.has_more) return campaigns;
    offset += data.count;
  }
}

// RFC 4180: quote when needed, double the quotes, null and undefined become empty.
const cell = (value) => {
  if (value === null || value === undefined) return "";
  const text = typeof value === "object" ? JSON.stringify(value) : String(value);
  return /[",\n\r]/.test(text) ? `"${text.replace(/"/g, '""')}"` : text;
};
const csv = (columns, rows) => [columns.join(","), ...rows.map((row) => columns.map((column) => cell(row[column])).join(","))].join("\n") + "\n";

const DAILY_COLUMNS = ["exported_at", "campaign_id", "campaign_name", "day", "sender_key", "variation", "platform", "partial", "emails_sent", "emails_opened", "li_connection_requests", "li_connections_accepted", "li_messages", "leads_first_contacted", "leads_first_replied", "leads_first_positive", "replies_positive", "replies_negative", "replies_neutral", "replies_ooo", "meetings_booked"];
const SENDER_COLUMNS = ["exported_at", "window", "campaign_id", "campaign_name", "sender_key", "account_id", "label", "platform", "is_active", "messages_sent", "requests", "accepted", "acceptance_rate_pct", "contacted_leads", "replied_leads", "reply_rate_pct", "positive_leads", "positive_rate_pct"];
const STEP_COLUMNS = ["exported_at", "window", "campaign_id", "campaign_name", "step_id", "position", "lane", "type", "channel", "day", "subject", "is_message_step", "reached", "waiting", "replied", "reply_rate_pct", "positive", "positive_rate_pct"];
const ANALYSIS_COLUMNS = ["exported_at", "window", "campaign_id", "campaign_name", "is_active", "enrolled", "not_contacted", "active", "paused", "completed", "contacted", "replied", "positive", "meetings", "reply_rate_pct", "positive_rate_pct", "meeting_rate_pct", "replies_total", "replies_negative", "replies_neutral", "replies_ooo"];

const exportedAt = new Date().toISOString();
const today = exportedAt.slice(0, 10);
const from = new Date(Date.now() - DAYS * 24 * 60 * 60 * 1000).toISOString().slice(0, 10);
const tables = { daily_stats_by_sender: [], daily_stats_by_variation: [], sender_breakdown: [], step_funnel: [], campaign_analysis: [] };

for (const campaign of await listCampaigns()) {
  const tag = { exported_at: exportedAt, campaign_id: campaign.id, campaign_name: campaign.name };
  for (const [table, groupBy] of [["daily_stats_by_sender", "sender"], ["daily_stats_by_variation", "variation"]]) {
    const stats = await call(`/campaigns/${campaign.id}/daily-stats?from=${from}&to=${today}&group_by=${groupBy}`);
    tables[table].push(...stats.days.map((row) => ({ ...tag, ...row })));
  }
  const senders = await call(`/campaigns/${campaign.id}/senders/breakdown?date_filter=${WINDOW}`);
  tables.sender_breakdown.push(...senders.senders.map((row) => ({ ...tag, window: WINDOW, ...row })));
  const funnel = await call(`/campaigns/${campaign.id}/step-funnel?date_filter=${WINDOW}`);
  tables.step_funnel.push(...funnel.steps.map((row) => ({ ...tag, window: WINDOW, ...row })));
  const { analysis } = await call(`/campaigns/${campaign.id}/analysis?date_filter=${WINDOW}`);
  if (analysis) {
    const { leads = {}, funnel: f = {}, replies = {} } = analysis;
    tables.campaign_analysis.push({
      ...tag, window: WINDOW, is_active: campaign.is_active,
      enrolled: leads.enrolled, not_contacted: leads.not_contacted, active: leads.active, paused: leads.paused, completed: leads.completed,
      contacted: f.contacted, replied: f.replied, positive: f.positive, meetings: f.meetings,
      reply_rate_pct: f.reply_rate_pct, positive_rate_pct: f.positive_rate_pct, meeting_rate_pct: f.meeting_rate_pct,
      replies_total: replies.total, replies_negative: replies.negative, replies_neutral: replies.neutral, replies_ooo: replies.ooo,
    });
  }
  await sleep(650); // five calls per campaign, under 100 a minute to any one endpoint
}

await mkdir(OUT, { recursive: true });
const columns = { daily_stats_by_sender: DAILY_COLUMNS, daily_stats_by_variation: DAILY_COLUMNS, sender_breakdown: SENDER_COLUMNS, step_funnel: STEP_COLUMNS, campaign_analysis: ANALYSIS_COLUMNS };
for (const [table, rows] of Object.entries(tables)) {
  await writeFile(path.join(OUT, `${table}.csv`), csv(columns[table], rows));
  console.log(`${table}.csv: ${rows.length} row(s)`);
}
```

**Python**

```python
# export_analytics.py
import csv
import os
import time
from datetime import datetime, timedelta, timezone

import requests

API_URL = "https://api.versionseven.ai/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['VICTORIA_API_KEY']}"}
DAYS = int(os.environ.get("DAYS", 3))  # trailing days to re-pull for the daily tables
WINDOW = os.environ.get("WINDOW", "30d")  # date_filter for the snapshot tables
OUT = os.environ.get("OUT", "export")


def call(path):
    for attempt in range(1, 6):
        response = requests.get(f"{API_URL}{path}", headers=HEADERS, timeout=60)
        try:
            data = response.json()
        except ValueError:
            data = {}
        if response.status_code not in (429, 503) or attempt == 5:
            if response.status_code != 200:
                raise SystemExit(f"{path} answered {response.status_code} {data.get('error', '')}")
            return data
        time.sleep(float(response.headers.get("Retry-After") or 2**attempt))


def list_campaigns():
    campaigns, offset = [], 0
    while True:
        data = call(f"/campaigns?limit=500&offset={offset}")
        campaigns.extend(data["campaigns"])
        if not data["has_more"]:
            return campaigns
        offset += data["count"]


DAILY_COLUMNS = ["exported_at", "campaign_id", "campaign_name", "day", "sender_key", "variation", "platform", "partial", "emails_sent", "emails_opened", "li_connection_requests", "li_connections_accepted", "li_messages", "leads_first_contacted", "leads_first_replied", "leads_first_positive", "replies_positive", "replies_negative", "replies_neutral", "replies_ooo", "meetings_booked"]
SENDER_COLUMNS = ["exported_at", "window", "campaign_id", "campaign_name", "sender_key", "account_id", "label", "platform", "is_active", "messages_sent", "requests", "accepted", "acceptance_rate_pct", "contacted_leads", "replied_leads", "reply_rate_pct", "positive_leads", "positive_rate_pct"]
STEP_COLUMNS = ["exported_at", "window", "campaign_id", "campaign_name", "step_id", "position", "lane", "type", "channel", "day", "subject", "is_message_step", "reached", "waiting", "replied", "reply_rate_pct", "positive", "positive_rate_pct"]
ANALYSIS_COLUMNS = ["exported_at", "window", "campaign_id", "campaign_name", "is_active", "enrolled", "not_contacted", "active", "paused", "completed", "contacted", "replied", "positive", "meetings", "reply_rate_pct", "positive_rate_pct", "meeting_rate_pct", "replies_total", "replies_negative", "replies_neutral", "replies_ooo"]
COLUMNS = {"daily_stats_by_sender": DAILY_COLUMNS, "daily_stats_by_variation": DAILY_COLUMNS, "sender_breakdown": SENDER_COLUMNS, "step_funnel": STEP_COLUMNS, "campaign_analysis": ANALYSIS_COLUMNS}


def main():
    exported_at = datetime.now(timezone.utc).isoformat()
    today = exported_at[:10]
    start = (datetime.now(timezone.utc) - timedelta(days=DAYS)).date().isoformat()
    tables = {name: [] for name in COLUMNS}

    for campaign in list_campaigns():
        tag = {"exported_at": exported_at, "campaign_id": campaign["id"], "campaign_name": campaign["name"]}
        for table, group_by in (("daily_stats_by_sender", "sender"), ("daily_stats_by_variation", "variation")):
            stats = call(f"/campaigns/{campaign['id']}/daily-stats?from={start}&to={today}&group_by={group_by}")
            tables[table].extend({**tag, **row} for row in stats["days"])
        senders = call(f"/campaigns/{campaign['id']}/senders/breakdown?date_filter={WINDOW}")
        tables["sender_breakdown"].extend({**tag, "window": WINDOW, **row} for row in senders["senders"])
        funnel = call(f"/campaigns/{campaign['id']}/step-funnel?date_filter={WINDOW}")
        tables["step_funnel"].extend({**tag, "window": WINDOW, **row} for row in funnel["steps"])
        analysis = call(f"/campaigns/{campaign['id']}/analysis?date_filter={WINDOW}").get("analysis")
        if analysis:
            leads, f, replies = analysis.get("leads") or {}, analysis.get("funnel") or {}, analysis.get("replies") or {}
            tables["campaign_analysis"].append({
                **tag, "window": WINDOW, "is_active": campaign.get("is_active"),
                "enrolled": leads.get("enrolled"), "not_contacted": leads.get("not_contacted"), "active": leads.get("active"), "paused": leads.get("paused"), "completed": leads.get("completed"),
                "contacted": f.get("contacted"), "replied": f.get("replied"), "positive": f.get("positive"), "meetings": f.get("meetings"),
                "reply_rate_pct": f.get("reply_rate_pct"), "positive_rate_pct": f.get("positive_rate_pct"), "meeting_rate_pct": f.get("meeting_rate_pct"),
                "replies_total": replies.get("total"), "replies_negative": replies.get("negative"), "replies_neutral": replies.get("neutral"), "replies_ooo": replies.get("ooo"),
            })
        time.sleep(0.65)  # five calls per campaign, under 100 a minute to any one endpoint

    os.makedirs(OUT, exist_ok=True)
    for table, rows in tables.items():
        with open(os.path.join(OUT, f"{table}.csv"), "w", newline="", encoding="utf-8") as handle:
            writer = csv.DictWriter(handle, fieldnames=COLUMNS[table], extrasaction="ignore")
            writer.writeheader()
            for row in rows:
                writer.writerow({key: ("" if value is None else "true" if value is True else "false" if value is False else value) for key, value in row.items()})
        print(f"{table}.csv: {len(rows)} row(s)")


if __name__ == "__main__":
    main()
```

```bash
node export-analytics.mjs             # writes export/*.csv for the trailing 3 days and a 30d window
DAYS=90 node export-analytics.mjs     # first run: backfill 90 days of the daily tables
python export_analytics.py
```

Values in both CSVs are written the same way: `null` becomes an empty cell, booleans `true`/`false`, and text that contains a comma or quote is quoted.

## Loading into Postgres

The tables, with the merge keys as primary keys:

```sql
create table if not exists victoria_daily_stats_by_sender (
  campaign_id uuid not null, campaign_name text, day date not null, sender_key text not null default '',
  platform text, partial boolean, exported_at timestamptz,
  emails_sent int, emails_opened int, li_connection_requests int, li_connections_accepted int, li_messages int,
  leads_first_contacted int, leads_first_replied int, leads_first_positive int,
  replies_positive int, replies_negative int, replies_neutral int, replies_ooo int, meetings_booked int,
  primary key (campaign_id, day, sender_key)
);
create table if not exists victoria_daily_stats_by_variation (like victoria_daily_stats_by_sender including all);
alter table victoria_daily_stats_by_variation drop constraint victoria_daily_stats_by_variation_pkey,
  add column if not exists variation text not null default '',
  add primary key (campaign_id, day, variation);
create table if not exists victoria_campaign_analysis (
  exported_at timestamptz not null, "window" text, campaign_id uuid not null, campaign_name text, is_active boolean,
  enrolled int, not_contacted int, active int, paused int, completed int,
  contacted int, replied int, positive int, meetings int,
  reply_rate_pct numeric, positive_rate_pct numeric, meeting_rate_pct numeric,
  replies_total int, replies_negative int, replies_neutral int, replies_ooo int,
  primary key (exported_at, campaign_id)
);
```

Load each daily CSV through a staging table and merge on the key, so a re-pulled day updates in place:

```sql
create temp table staging (like victoria_daily_stats_by_sender including defaults);
\copy staging (exported_at, campaign_id, campaign_name, day, sender_key, variation, platform, partial, emails_sent, emails_opened, li_connection_requests, li_connections_accepted, li_messages, leads_first_contacted, leads_first_replied, leads_first_positive, replies_positive, replies_negative, replies_neutral, replies_ooo, meetings_booked) from 'export/daily_stats_by_sender.csv' with (format csv, header true, null '');
insert into victoria_daily_stats_by_sender
  select exported_at, campaign_id, campaign_name, day, coalesce(sender_key, ''), platform, partial, emails_sent, emails_opened, li_connection_requests, li_connections_accepted, li_messages, leads_first_contacted, leads_first_replied, leads_first_positive, replies_positive, replies_negative, replies_neutral, replies_ooo, meetings_booked
  from staging
on conflict (campaign_id, day, sender_key) do update set
  exported_at = excluded.exported_at, partial = excluded.partial,
  emails_sent = excluded.emails_sent, emails_opened = excluded.emails_opened,
  li_connection_requests = excluded.li_connection_requests, li_connections_accepted = excluded.li_connections_accepted, li_messages = excluded.li_messages,
  leads_first_contacted = excluded.leads_first_contacted, leads_first_replied = excluded.leads_first_replied, leads_first_positive = excluded.leads_first_positive,
  replies_positive = excluded.replies_positive, replies_negative = excluded.replies_negative, replies_neutral = excluded.replies_neutral, replies_ooo = excluded.replies_ooo,
  meetings_booked = excluded.meetings_booked;
```

The staging table needs a `variation` column for the `\copy` column list; add it with `alter table staging add column variation text` before the copy, or copy the variation CSV into a staging table created from the variation table instead. The snapshot CSVs (`sender_breakdown`, `step_funnel`, `campaign_analysis`) append: their key includes `exported_at`, so a plain `\copy` into the table is enough and the latest `exported_at` is the current snapshot.

## Loading into BigQuery

```bash
bq load --source_format=CSV --skip_leading_rows=1 --autodetect outbound.daily_stats_by_sender_staging export/daily_stats_by_sender.csv
bq query --use_legacy_sql=false '
merge outbound.daily_stats_by_sender t
using outbound.daily_stats_by_sender_staging s
on t.campaign_id = s.campaign_id and t.day = s.day and ifnull(t.sender_key, "") = ifnull(s.sender_key, "")
when matched then update set exported_at = s.exported_at, partial = s.partial, emails_sent = s.emails_sent, leads_first_contacted = s.leads_first_contacted, leads_first_replied = s.leads_first_replied, replies_positive = s.replies_positive, meetings_booked = s.meetings_booked
when not matched then insert row'
```

List every column in the `update set` the first time; the pattern is the same as Postgres.

## Schedule it

Nightly after midnight UTC is right: the daily rows for yesterday are complete by then, and `DAYS=3` catches the few that settle late. The crontab or GitHub Actions pattern from [the daily digest](https://docs.versionseven.ai/cookbook/daily-campaign-digest#schedule-it) applies, followed by the load step. On the first run, set `DAYS` to as far back as you want history; the daily endpoints accept any range.

## What to expect

- **`partial` is true for today's row** in the daily tables, and only today's. Charts should filter it out or shade it.
- **Grouped rows carry their grain**: `sender_key` is set and `variation` empty in the by-sender file, and the reverse in the by-variation file. A campaign without an A/B test still has a `variation` of `a` or `unassigned` per the app's rule that A + B + unassigned equals all.
- **Rates are `null` when their denominator is zero**, which becomes an empty cell. Recompute rates in SQL from the counts rather than averaging the percentages.
- **Meetings** are leads the Appointment Setter or a person confirmed as booked, not calendar events, unless Calendly is connected. [Analytics metrics](https://docs.versionseven.ai/help/analytics-metrics) says so for every metric.
- Five calls per campaign per night: a hundred campaigns is five hundred requests, inside the limits with the pause between campaigns.

## Next steps

- [Post a daily campaign digest to Slack](https://docs.versionseven.ai/cookbook/daily-campaign-digest) for the same numbers as a message rather than a table.
- [Run agency operations across client workspaces](https://docs.versionseven.ai/cookbook/agency-operations) to export every client's workspace into one warehouse.
- [Analytics metrics](https://docs.versionseven.ai/help/analytics-metrics) for the definition of each column.