Cookbook
View as Markdown

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.

Last updated

  • Postgres
  • BigQuery
  • Metabase
  • cron

Endpoints used

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.
  • 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

FileSourceGrainMerge key
daily_stats_by_sender.csvGET /v1/campaigns/{campaign_id}/daily-stats with group_by=sendercampaign × day × sendercampaign_id, day, sender_key
daily_stats_by_variation.csvthe same with group_by=variationcampaign × day × A/B armcampaign_id, day, variation
sender_breakdown.csvGET /v1/campaigns/{campaign_id}/senders/breakdowncampaign × sender, over the windowexported_at, campaign_id, sender_key
step_funnel.csvGET /v1/campaigns/{campaign_id}/step-funnelcampaign × step, over the windowexported_at, campaign_id, step_id
campaign_analysis.csvGET /v1/campaigns/{campaign_id}/analysiscampaign, over the windowexported_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 defines each column.

The script

// 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)`);
}
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:

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:

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

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 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 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