import os, requests, time
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"}
PERSONA_IDS = os.environ.get("PERSONA_IDS", "").split(",")
# 1. Find completed A/B test single sends
sends = requests.get(f"{SG_BASE}/marketing/singlesends",
headers=SG_H, params={"page_size": 50}).json().get("result", [])
ab_sends = [s for s in sends
if s.get("status") == "triggered"
and s.get("send_to", {}).get("ab_test")]
if not ab_sends:
print("No completed A/B tests found")
exit()
latest = ab_sends[0]
send_id = latest["id"]
send_name = latest.get("name", "Untitled")
# 2. Get A/B test details and stats
detail = requests.get(f"{SG_BASE}/marketing/singlesends/{send_id}",
headers=SG_H).json()
ab_config = detail.get("send_to", {}).get("ab_test", {})
stats = requests.get(f"{SG_BASE}/marketing/stats/singlesends/{send_id}",
headers=SG_H, params={"aggregated_by": "total"}).json()
# 3. Determine winner/loser
ab_phase = ab_config.get("ab_phase", "send")
winner_id = ab_config.get("winner_id", "")
variant_a = {
"subject": detail.get("subject", ""),
"html_content": detail.get("html_content", "")[:800],
"plain_content": detail.get("plain_content", "")[:500],
}
# Pull variant B subject from A/B config
variant_b_subject = ""
for v in ab_config.get("variants", []):
if v.get("id") != winner_id:
variant_b_subject = v.get("subject", "")
break
loser_subject = variant_b_subject or variant_a["subject"]
loser_body = variant_a["plain_content"] or "Email body not available"
print(f"A/B Test: {send_name}")
print(f"Winner: {variant_a['subject']}")
print(f"Loser: {loser_subject}")
# 4. Run Focus Group on losing variant
if not PERSONA_IDS or PERSONA_IDS == [""]:
persona = requests.post(f"{MB}/personas", headers=MV_H, json={
"name": "SG A/B Test Audience",
"description": "General email subscriber. Mix of B2B decision makers and practitioners.",
}).json()
PERSONA_IDS = [persona["id"]]
fg = requests.post(f"{MB}/focus-groups", headers=MV_H, json={
"name": f"A/B Extension: {send_name}",
"persona_ids": PERSONA_IDS,
"context": f"""This email variant LOST an A/B test against a competitor variant.
LOSING VARIANT:
Subject: {loser_subject}
Body: {loser_body}
WINNING VARIANT (for reference):
Subject: {variant_a['subject']}
The winning variant had a higher open rate. We want to understand WHY
the losing variant underperformed and how to improve it.""",
"questions": [
"Read the losing subject line. What is your immediate reaction? Would you open this?",
"What specific words or phrases weaken this subject line?",
"What would need to change about this email for you to click through?",
"Rewrite the losing subject line in a way you would actually open.",
"Compare both subject lines. What makes the winner more compelling?",
],
"responses_per_persona": 3,
}).json()
# 5. Poll for results
for _ in range(20):
time.sleep(5)
data = requests.get(f"{MB}/focus-groups/{fg['id']}",
headers=MV_H).json()
if data.get("status") == "completed":
break
print(f"\nFocus Group: {fg['id']}")
for resp in data.get("responses", []):
print(f"\n[{resp.get('persona_name', '?')}] {resp.get('question', '')[:60]}")
print(f" → {resp.get('answer', '')[:200]}")