#!/usr/bin/env python3 """ Klaviyo Data Fetcher for ASPEKT Dashboard Fetches real-time data from all 10 clients via Klaviyo API Outputs JSON for dashboard consumption """ import json import urllib.request import urllib.error import urllib.parse from datetime import datetime, timedelta import sys import os # Klaviyo client configurations CLIENTS = { "Kompsos Ltd": { "api_key": "pk_1ae14f2c7bb0f5881d8421a591ee0465db", "currency": "GBP" }, "Til Valhalla Project": { "api_key": "pk_ac7ce0699d96e47c9b118c057568d1040d", "currency": "USD" }, "Caviar Star": { "api_key": "pk_bb45a70ad3a2f0ae8d6c135d75144c81a3", "currency": "USD" }, "Crafty By Numbers": { "api_key": "pk_bb8ff58f4c7641e5660bbc9be5150ddf37", "currency": "USD" }, "Dingo1969": { "api_key": "pk_18a21d4249fea96f1a3a3d5e08e937bfc8", "currency": "USD" }, "Marigold": { "api_key": "pk_b9f15d69f9d2d5f1f899b849bfe3856cac", "currency": "USD" }, "Plum Paper": { "api_key": "pk_8f44670613a471f027e4418bf2b8082478", "currency": "USD" }, "Serious Grit": { "api_key": "pk_52766c7e7b5c6a6c93d2c5ab953ee16070", "currency": "USD" }, "SHITI Coolers": { "api_key": "pk_1451a78e99ca823cd665737751fd68cfc2", "currency": "USD" }, "VAbody": { "api_key": "pk_e86c718af64f2ad96c297329e30b752510", "currency": "USD" } } BASE_URL = "https://a.klaviyo.com/api" API_REVISION = "2024-02-15" def make_klaviyo_request(endpoint, api_key, params=None): """Make authenticated request to Klaviyo API""" url = f"{BASE_URL}/{endpoint}" if params: url += "?" + urllib.parse.urlencode(params) headers = { "Authorization": f"Klaviyo-API-Key {api_key}", "Accept": "application/json", "Revision": API_REVISION } try: req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=30) as response: return json.loads(response.read().decode()) except urllib.error.HTTPError as e: print(f"HTTP Error {e.code}: {e.reason} for {url}") return None except urllib.error.URLError as e: print(f"URL Error: {e.reason} for {url}") return None except Exception as e: print(f"Unexpected error: {e} for {url}") return None def get_campaigns_data(api_key, days=30): """Get campaign data for last N days""" end_date = datetime.now() start_date = end_date - timedelta(days=days) params = { "filter": f"greater-than(send_time,{start_date.isoformat()})," + f"less-than(send_time,{end_date.isoformat()})", "include": "campaign-messages", "page[size]": "100" } campaigns = make_klaviyo_request("campaigns", api_key, params) if not campaigns: return [] campaign_data = [] for campaign in campaigns.get("data", []): attrs = campaign.get("attributes", {}) # Get campaign stats campaign_id = campaign.get("id") stats_params = {"page[size]": "100"} stats = make_klaviyo_request(f"campaign-messages/{campaign_id}/campaign-message-report-url", api_key, stats_params) campaign_data.append({ "id": campaign_id, "name": attrs.get("name", "Unknown"), "status": attrs.get("status", "unknown"), "send_time": attrs.get("send_time"), "subject": attrs.get("subject_line", ""), "recipients": attrs.get("sent_at", 0), "stats": stats if stats else {} }) return campaign_data def get_flows_data(api_key): """Get flow data""" flows = make_klaviyo_request("flows", api_key, {"page[size]": "100"}) if not flows: return [] flow_data = [] for flow in flows.get("data", []): attrs = flow.get("attributes", {}) flow_data.append({ "id": flow.get("id"), "name": attrs.get("name", "Unknown"), "status": attrs.get("status", "unknown"), "trigger_type": attrs.get("trigger_type", "unknown"), "created": attrs.get("created"), "updated": attrs.get("updated") }) return flow_data def get_metrics_data(api_key, days=30): """Get aggregated metrics for dashboard KPIs""" end_date = datetime.now() start_date = end_date - timedelta(days=days) # Get campaign metrics params = { "start_date": start_date.strftime("%Y-%m-%d"), "end_date": end_date.strftime("%Y-%m-%d"), "metric_name": ["campaign.received", "campaign.opened_email", "campaign.clicked_email", "campaign.placed_order"], "page[size]": "100" } metrics = make_klaviyo_request("metric-aggregates", api_key, params) # Default metrics structure return { "recipients": 0, "opens": 0, "clicks": 0, "orders": 0, "revenue": 0, "open_rate": 0, "click_rate": 0, "conversion_rate": 0 } def get_lists_data(api_key): """Get list data for subscriber counts""" lists = make_klaviyo_request("lists", api_key, {"page[size]": "100"}) if not lists: return [] list_data = [] for lst in lists.get("data", []): attrs = lst.get("attributes", {}) list_data.append({ "id": lst.get("id"), "name": attrs.get("name", "Unknown"), "profile_count": attrs.get("profile_count", 0), "created": attrs.get("created"), "updated": attrs.get("updated") }) return list_data def fetch_client_data(client_name, config): """Fetch all data for a single client""" print(f"Fetching data for {client_name}...") api_key = config["api_key"] currency = config.get("currency", "USD") # Fetch all data types campaigns = get_campaigns_data(api_key, 30) flows = get_flows_data(api_key) metrics = get_metrics_data(api_key, 30) lists = get_lists_data(api_key) return { "name": client_name, "currency": currency, "last_updated": datetime.now().isoformat(), "campaigns": campaigns, "flows": flows, "metrics": metrics, "lists": lists } def calculate_aggregated_kpis(all_clients_data): """Calculate aggregated KPIs across all clients""" total_recipients = 0 total_opens = 0 total_clicks = 0 total_orders = 0 total_revenue = 0 for client_data in all_clients_data: metrics = client_data.get("metrics", {}) total_recipients += metrics.get("recipients", 0) total_opens += metrics.get("opens", 0) total_clicks += metrics.get("clicks", 0) total_orders += metrics.get("orders", 0) total_revenue += metrics.get("revenue", 0) # Calculate rates avg_open_rate = (total_opens / total_recipients * 100) if total_recipients > 0 else 0 avg_click_rate = (total_clicks / total_recipients * 100) if total_recipients > 0 else 0 avg_conversion_rate = (total_orders / total_recipients * 100) if total_recipients > 0 else 0 avg_order_value = (total_revenue / total_orders) if total_orders > 0 else 0 return { "recipients": total_recipients, "avg_open_rate": round(avg_open_rate, 1), "avg_click_rate": round(avg_click_rate, 1), "avg_conversion_rate": round(avg_conversion_rate, 1), "avg_order_value": round(avg_order_value, 2), "total_revenue": round(total_revenue, 2), "total_orders": total_orders } def main(): """Main execution function""" print("Starting Klaviyo data fetch...") print(f"Timestamp: {datetime.now().isoformat()}") all_clients_data = [] # Fetch data for each client for client_name, config in CLIENTS.items(): try: client_data = fetch_client_data(client_name, config) all_clients_data.append(client_data) except Exception as e: print(f"Error fetching data for {client_name}: {e}") # Add placeholder data to prevent dashboard from breaking all_clients_data.append({ "name": client_name, "currency": config.get("currency", "USD"), "last_updated": datetime.now().isoformat(), "error": str(e), "campaigns": [], "flows": [], "metrics": { "recipients": 0, "opens": 0, "clicks": 0, "orders": 0, "revenue": 0, "open_rate": 0, "click_rate": 0, "conversion_rate": 0 }, "lists": [] }) # Calculate aggregated KPIs aggregated_kpis = calculate_aggregated_kpis(all_clients_data) # Prepare final output output_data = { "generated_at": datetime.now().isoformat(), "clients": all_clients_data, "aggregated_kpis": aggregated_kpis, "total_clients": len(all_clients_data) } # Write to JSON file output_file = "/Users/neoclaw/.openclaw/workspace/tools/klaviyo-dashboard-data.json" try: with open(output_file, 'w') as f: json.dump(output_data, f, indent=2, default=str) print(f"Data successfully written to {output_file}") print(f"Total clients processed: {len(all_clients_data)}") print(f"Aggregated KPIs: {aggregated_kpis}") except Exception as e: print(f"Error writing output file: {e}") sys.exit(1) if __name__ == "__main__": main()