import os, requests, time
from collections import defaultdict
from datetime import datetime, timedelta
SG = os.environ["SENDGRID_API_KEY"]
MV = os.environ["MAVERA_API_KEY"]
SG_BASE = "https://api.sendgrid.com/v3"
MB = "https://app.mavera.io/api/v1"
SG_H = {"Authorization": f"Bearer {SG}"}
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
# 1. List recent single sends
sends = requests.get(f"{SG_BASE}/marketing/singlesends",
headers=SG_H, params={"page_size": 25}).json().get("result", [])
completed = [s for s in sends if s.get("status") == "triggered"]
# 2. Pull stats with breakdowns
device_geo = defaultdict(lambda: {"opens": 0, "clicks": 0, "sends": 0})
for send in completed[:15]:
sid = send["id"]
stats = requests.get(f"{SG_BASE}/marketing/stats/singlesends/{sid}",
headers=SG_H, params={
"aggregated_by": "total",
"start_date": (datetime.now() - timedelta(days=90)).strftime("%Y-%m-%d"),
"end_date": datetime.now().strftime("%Y-%m-%d"),
})
if stats.status_code == 429:
time.sleep(2)
stats = requests.get(f"{SG_BASE}/marketing/stats/singlesends/{sid}",
headers=SG_H, params={
"aggregated_by": "total",
"start_date": (datetime.now() - timedelta(days=90)).strftime("%Y-%m-%d"),
"end_date": datetime.now().strftime("%Y-%m-%d"),
})
stats.raise_for_status()
data = stats.json().get("results", [{}])
for entry in data:
metrics = entry.get("stats", {})
device_geo["all"]["opens"] += metrics.get("opens", 0)
device_geo["all"]["clicks"] += metrics.get("clicks", 0)
device_geo["all"]["sends"] += metrics.get("delivered", 0)
time.sleep(0.15)
# 3. Pull global stats with device/geo breakdowns
global_stats = requests.get(f"{SG_BASE}/stats",
headers=SG_H, params={
"start_date": (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d"),
"end_date": datetime.now().strftime("%Y-%m-%d"),
"aggregated_by": "month",
}).json()
browsers = requests.get(f"{SG_BASE}/browsers/stats",
headers=SG_H, params={
"start_date": (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d"),
"end_date": datetime.now().strftime("%Y-%m-%d"),
"aggregated_by": "month",
})
browser_data = browsers.json() if browsers.ok else []
# 4. Analyze patterns with Mave
stats_summary = f"""Email Performance (last 30 days):
Total sends: {device_geo['all']['sends']}
Total opens: {device_geo['all']['opens']}
Total clicks: {device_geo['all']['clicks']}
Open rate: {device_geo['all']['opens'] / max(device_geo['all']['sends'], 1):.1%}
CTR: {device_geo['all']['clicks'] / max(device_geo['all']['sends'], 1):.1%}
Browser breakdown: {browser_data[:3] if browser_data else 'N/A'}
Single sends analyzed: {len(completed[:15])}"""
analysis = requests.post(f"{MB}/mave/chat", headers=MV_H, json={
"message": f"""Analyze this SendGrid email performance data and produce persona
behavioral profiles.
{stats_summary}
For each inferred segment, describe:
1) Likely device preference (mobile vs desktop)
2) Content format preference (scannable vs detailed)
3) Optimal send time window
4) Subject line style that works
5) CTA format preference
Group into 3 behavioral personas: Mobile-First, Desktop-Focused, Hybrid."""
}).json()
print("=== Behavioral Analysis ===")
print(analysis.get("content", "")[:1500])
# 5. Create behavioral personas
BEHAVIOR_PROFILES = [
{"name": "SG: Mobile-First Reader",
"desc": "Opens on mobile (60%+). Prefers scannable content, single CTA, short subject lines.",
"psych": {"device": "mobile", "format": "scannable", "cta": "single_button"}},
{"name": "SG: Desktop Focused Reader",
"desc": "Opens on desktop during business hours. Prefers detailed content, multiple links, professional tone.",
"psych": {"device": "desktop", "format": "detailed", "cta": "text_links"}},
{"name": "SG: Hybrid Multi-Device",
"desc": "Mixed device usage. Opens on mobile, clicks through on desktop. Needs responsive design.",
"psych": {"device": "hybrid", "format": "responsive", "cta": "progressive"}},
]
for profile in BEHAVIOR_PROFILES:
r = requests.post(f"{MB}/personas", headers=MV_H, json={
"name": profile["name"],
"description": f"{profile['desc']} Based on {device_geo['all']['sends']} sends analyzed. {analysis.get('content', '')[:200]}",
"psychographic": profile["psych"],
})
r.raise_for_status()
print(f"Created: {profile['name']} → {r.json()['id']}")
time.sleep(0.3)
