#!/usr/bin/env python3 """Klaviyo 90-day email + SMS audit fetch. Usage: KLAVIYO_KEY=pk_live_xxxxxxxxxxxx python3 klaviyo_fetch.py Pulls: - Email campaigns (last 90d) → klaviyo/email_campaigns.json + .csv - SMS campaigns → klaviyo/sms_campaigns.json + .csv - Per-campaign metrics → recipients, opens, clicks, conversions, revenue - Flow performance → klaviyo/flows.json - Audience growth (lists) → klaviyo/lists.json - Top segments → klaviyo/segments.json Klaviyo private keys start with `pk_` and are 50+ chars. """ import os, sys, json, time, requests from datetime import datetime, timedelta KEY = os.environ.get("KLAVIYO_KEY") or (sys.argv[1] if len(sys.argv) > 1 else "") if not KEY: print("ERROR: pass Klaviyo private key as KLAVIYO_KEY env var or first arg", file=sys.stderr) sys.exit(2) OUT = "/home/sol1/Desktop/MARKETING/nitrous_3mo/klaviyo" os.makedirs(OUT, exist_ok=True) H = { "Authorization": f"Klaviyo-API-Key {KEY}", "accept": "application/vnd.api+json", "revision": "2024-10-15", } START = (datetime.utcnow() - timedelta(days=90)).strftime("%Y-%m-%dT00:00:00Z") END = datetime.utcnow().strftime("%Y-%m-%dT23:59:59Z") print(f"Window: {START} -> {END}", flush=True) def get(url, params=None, retries=4): for i in range(retries): r = requests.get(url, headers=H, params=params, timeout=60) if r.status_code == 200: return r.json() if r.status_code == 429: time.sleep(2 + i*2); continue print(f"GET {url} -> {r.status_code} {r.text[:300]}", flush=True) if r.status_code in (401, 403): sys.exit(3) time.sleep(1 + i) r.raise_for_status() def page_all(url, params=None): """Generic paginator for Klaviyo's JSON:API cursor format.""" out = [] while url: d = get(url, params=params) out.extend(d.get("data", []) or []) url = (d.get("links") or {}).get("next") params = None # next link already encodes params return out # -------- 1. Email campaigns -------- print("Fetching email campaigns...", flush=True) email_url = "https://a.klaviyo.com/api/campaigns" email_filter = f"and(equals(messages.channel,'email'),greater-or-equal(scheduled_at,{START}))" email_campaigns = page_all(email_url, params={"filter": email_filter}) print(f" {len(email_campaigns)} email campaigns", flush=True) with open(f"{OUT}/email_campaigns.json","w") as f: json.dump(email_campaigns, f) # -------- 2. SMS campaigns -------- print("Fetching SMS campaigns...", flush=True) sms_filter = f"and(equals(messages.channel,'sms'),greater-or-equal(scheduled_at,{START}))" sms_campaigns = page_all(email_url, params={"filter": sms_filter}) print(f" {len(sms_campaigns)} SMS campaigns", flush=True) with open(f"{OUT}/sms_campaigns.json","w") as f: json.dump(sms_campaigns, f) # -------- 3. Campaign performance via Reporting API -------- def campaign_metrics(campaign_ids, conv_metric_id=None): if not campaign_ids: return [] # Klaviyo Campaign Values endpoint metrics = [] # batch in 50s and back off on 429 for i in range(0, len(campaign_ids), 50): chunk = campaign_ids[i:i+50] body = { "data": { "type": "campaign-values-report", "attributes": { "statistics": [ "recipients","opens","opens_unique","open_rate", "clicks","clicks_unique","click_rate","click_to_open_rate", "conversions","conversion_uniques","conversion_value","conversion_rate", "delivered","delivery_rate","unsubscribes","unsubscribe_rate", "spam_complaints","spam_complaint_rate","bounced","bounce_rate", "revenue_per_recipient" ], "timeframe": {"key": "last_90_days"}, "filter": f"contains-any(campaign_id,[{','.join(repr(c) for c in chunk)}])", "conversion_metric_id": conv_metric_id, } } } for attempt in range(6): r = requests.post( "https://a.klaviyo.com/api/campaign-values-reports/", headers={**H, "content-type":"application/vnd.api+json"}, json=body, timeout=60, ) if r.status_code in (200, 207): metrics.append(r.json()) break if r.status_code == 429: wait = 3 + attempt*3 print(f" rate-limited, sleeping {wait}s...", flush=True) time.sleep(wait); continue print(f" metrics batch HTTP {r.status_code}: {r.text[:300]}") break time.sleep(1) # be polite return metrics # -------- 3a. Resolve "Placed Order" metric ID -------- print("Resolving Placed Order metric...", flush=True) metrics_data = page_all("https://a.klaviyo.com/api/metrics", params=None) po_id = None for m in metrics_data: nm = (m.get("attributes",{}).get("name") or "").lower() if nm == "placed order": po_id = m["id"]; break print(f" Placed Order metric id: {po_id}", flush=True) with open(f"{OUT}/metrics.json","w") as f: json.dump(metrics_data, f) # -------- 3b. Pull metrics for both channels -------- if po_id: print("Fetching email campaign performance...", flush=True) em_ids = [c["id"] for c in email_campaigns] em_perf = campaign_metrics(em_ids, conv_metric_id=po_id) with open(f"{OUT}/email_campaign_metrics.json","w") as f: json.dump(em_perf, f) print(f" saved {len(em_perf)} email metric batches", flush=True) print("Fetching SMS campaign performance...", flush=True) sms_ids = [c["id"] for c in sms_campaigns] sm_perf = campaign_metrics(sms_ids, conv_metric_id=po_id) with open(f"{OUT}/sms_campaign_metrics.json","w") as f: json.dump(sm_perf, f) print(f" saved {len(sm_perf)} sms metric batches", flush=True) # -------- 4. Flows -------- print("Fetching flows...", flush=True) flows = page_all("https://a.klaviyo.com/api/flows") with open(f"{OUT}/flows.json","w") as f: json.dump(flows, f) print(f" {len(flows)} flows", flush=True) # -------- 5. Lists & segments -------- print("Fetching lists...", flush=True) lists = page_all("https://a.klaviyo.com/api/lists", params=None) with open(f"{OUT}/lists.json","w") as f: json.dump(lists, f) print(f" {len(lists)} lists", flush=True) print("Fetching segments...", flush=True) segments = page_all("https://a.klaviyo.com/api/segments", params=None) with open(f"{OUT}/segments.json","w") as f: json.dump(segments, f) print(f" {len(segments)} segments", flush=True) # -------- 6. Account info (currency etc) -------- acc = get("https://a.klaviyo.com/api/accounts") with open(f"{OUT}/account.json","w") as f: json.dump(acc, f) print("DONE", flush=True)