import os, requests, time
MC_KEY = os.environ["MAILCHIMP_API_KEY"]
MC_DC = os.environ["MAILCHIMP_DC"]
MV = os.environ["MAVERA_API_KEY"]
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
MC_BASE = f"https://{MC_DC}.api.mailchimp.com/3.0"
mc_auth = ("anystring", MC_KEY)
AB_CAMPAIGN_ID = os.environ.get("MAILCHIMP_AB_CAMPAIGN_ID", "")
if AB_CAMPAIGN_ID:
r = requests.get(f"{MC_BASE}/campaigns/{AB_CAMPAIGN_ID}", auth=mc_auth)
r.raise_for_status()
campaign = r.json()
ab_settings = campaign.get("variate_settings", {})
else:
ab_settings = {
"subject_lines": [
"Your Q1 numbers are in — here's what they mean",
"We analyzed your Q1 data (3 surprises inside)"
],
"winner_criteria": "open_rate",
}
campaign = {
"report_summary": {"open_rate": 0.32, "click_rate": 0.048},
}
subject_a = ab_settings.get("subject_lines", ["Subject A", "Subject B"])[0]
subject_b = ab_settings.get("subject_lines", ["Subject A", "Subject B"])[1] if len(ab_settings.get("subject_lines", [])) > 1 else "Subject B"
report_r = requests.get(f"{MC_BASE}/reports/{AB_CAMPAIGN_ID}/ab", auth=mc_auth) if AB_CAMPAIGN_ID else None
if report_r and report_r.status_code == 200:
ab_report = report_r.json()
a_stats = ab_report.get("a", {})
b_stats = ab_report.get("b", {})
else:
a_stats = {"open_rate": 0.28, "click_rate": 0.038, "unique_opens": 1400}
b_stats = {"open_rate": 0.36, "click_rate": 0.052, "unique_opens": 1800}
winner = "B" if b_stats.get("open_rate", 0) > a_stats.get("open_rate", 0) else "A"
PERSONA_IDS = os.environ.get("AB_PERSONA_IDS", "").split(",")
if not PERSONA_IDS[0]:
for name, desc in [
("Newsletter Regular", "Opens emails 2-3x per week. Skims subject lines quickly. Values clarity over cleverness."),
("Busy Executive", "Gets 200+ emails/day. Opens only what seems urgent or directly relevant. Reads on mobile."),
("Curious Explorer", "Opens most marketing emails. Enjoys discovering new ideas. Clicks through to articles."),
]:
p = requests.post(f"{MB}/personas", headers=MH, json={"name": name, "description": desc}).json()
PERSONA_IDS.append(p["id"])
time.sleep(0.3)
fg = requests.post(f"{MB}/focus-groups", headers=MH, json={
"name": "Mailchimp A/B Subject Line Investigation",
"persona_ids": [pid for pid in PERSONA_IDS if pid],
"questions": [
f'You see two emails in your inbox:\n A: "{subject_a}"\n B: "{subject_b}"\nWhich would you open first? Why? What about the winning subject line caught your attention?',
f'Subject A got {a_stats.get("open_rate", 0):.0%} opens and Subject B got {b_stats.get("open_rate", 0):.0%}. Does the winner ({winner}) surprise you? Why do you think it performed better?',
"What elements in a subject line make you most likely to open: (1) Personalization, (2) Curiosity gap, (3) Urgency, (4) Social proof, (5) Specificity (numbers/data)? Rank and explain your #1.",
"Write a subject line that would outperform BOTH of these. Explain your strategy.",
],
"context": f"""Mailchimp A/B test results:
VARIANT A: "{subject_a}"
Open rate: {a_stats.get('open_rate', 0):.0%} | Clicks: {a_stats.get('click_rate', 0):.0%} | Unique opens: {a_stats.get('unique_opens', 0):,}
VARIANT B: "{subject_b}"
Open rate: {b_stats.get('open_rate', 0):.0%} | Clicks: {b_stats.get('click_rate', 0):.0%} | Unique opens: {b_stats.get('unique_opens', 0):,}
Winner: Variant {winner} (by {ab_settings.get('winner_criteria', 'open_rate')})
Margin: {abs(a_stats.get('open_rate', 0) - b_stats.get('open_rate', 0)):.0%} difference""",
"responses_per_persona": 2,
}).json()
for _ in range(24):
time.sleep(5)
data = requests.get(f"{MB}/focus-groups/{fg['id']}", headers=MH).json()
if data.get("status") == "completed":
break
print(f"Focus Group: {data.get('id')} — {data.get('status')}\n")
for resp in data.get("responses", []):
print(f"[{resp.get('persona_id','?')}] {resp.get('question','')[:70]}")
print(f" → {resp.get('answer','')[:300]}\n")