A/B Copy Production

import os
import time
import json
import requests

API_KEY = os.environ["MAVERA_API_KEY"]
WORKSPACE_ID = os.environ["MAVERA_WORKSPACE_ID"]
BASE = "https://app.mavera.io/api/v1"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}


def create_voice(label, urls, usage_context):
    """Create a brand voice and wait for READY."""
    resp = requests.post(f"{BASE}/brand-voices", headers=HEADERS, json={
        "label": label,
        "usage_context": usage_context,
        "urls": urls,
        "workspace_id": WORKSPACE_ID,
    })
    resp.raise_for_status()
    bv = resp.json()

    for _ in range(30):
        check = requests.get(f"{BASE}/brand-voices/{bv['id']}", headers=HEADERS).json()
        if check.get("status") == "READY":
            return check
        time.sleep(10)
    raise TimeoutError(f"Brand voice {bv['id']} not ready")


VOICES = {
    "bold_direct": create_voice(
        label="Bold & Direct",
        urls=["https://yourbrand.com/blog", "https://yourbrand.com/manifesto"],
        usage_context="High-energy marketing copy that challenges the status quo",
    ),
    "warm_storytelling": create_voice(
        label="Warm & Storytelling",
        urls=["https://yourbrand.com/customer-stories", "https://yourbrand.com/about"],
        usage_context="Empathetic content that connects through customer narratives",
    ),
    "data_driven": create_voice(
        label="Data-Driven & Precise",
        urls=["https://yourbrand.com/research", "https://yourbrand.com/case-studies"],
        usage_context="Evidence-based content with metrics and benchmarks",
    ),
}

for name, voice in VOICES.items():
    print(f"  {name}: {voice['id']} — {voice.get('tone_adjectives', [])}")

SHARED_BRIEF = {
    "app_id": "email_sequence_generator",
    "input_data": {
        "sequence_goal": "Convert free trial users to paid plans",
        "emails_count": 3,
        "target_audience": "SaaS trial users in their first week",
        "key_points": [
            "Highlight the top 3 features they haven't tried",
            "Include a customer success story",
            "End with a time-limited upgrade offer",
        ],
    },
}


def wait_for_generation(gen_id, max_wait=300):
    for _ in range(max_wait // 10):
        resp = requests.get(f"{BASE}/generations/{gen_id}", headers=HEADERS)
        data = resp.json()
        if data.get("status") == "COMPLETED":
            return data
        time.sleep(10)
    raise TimeoutError(f"Generation {gen_id} timed out")


variants = {}
total_credits = 0

for voice_name, voice in VOICES.items():
    print(f"\nGenerating variant: {voice_name}...")

    payload = {
        **SHARED_BRIEF,
        "title": f"A/B Variant — {voice['label']}",
        "brand_voice_id": voice["id"],
        "workspace_id": WORKSPACE_ID,
    }

    resp = requests.post(f"{BASE}/generations", headers=HEADERS, json=payload)
    resp.raise_for_status()
    gen = resp.json()

    if gen.get("status") in ("PENDING", "RUNNING"):
        gen = wait_for_generation(gen["id"])

    credits = gen.get("usage", {}).get("credits_used", 0)
    total_credits += credits

    variants[voice_name] = {
        "voice_label": voice["label"],
        "voice_id": voice["id"],
        "output": gen.get("output", ""),
        "credits": credits,
    }
    print(f"  ✓ {credits} credits | {len(gen.get('output', ''))} chars")

print(f"\nGenerated {len(variants)} variants | {total_credits} total credits")

COMPARISON_SCHEMA = {
    "type": "json_schema",
    "json_schema": {
        "name": "ab_comparison",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "variants": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "tone_description": {"type": "string"},
                            "emotional_appeal": {"type": "number"},
                            "clarity": {"type": "number"},
                            "urgency": {"type": "number"},
                            "cta_strength": {"type": "number"},
                            "strengths": {"type": "array", "items": {"type": "string"}},
                            "weaknesses": {"type": "array", "items": {"type": "string"}},
                        },
                        "required": ["name", "tone_description", "emotional_appeal", "clarity", "urgency", "cta_strength", "strengths", "weaknesses"],
                    },
                },
                "recommendation": {"type": "string"},
                "test_hypothesis": {"type": "string"},
            },
            "required": ["variants", "recommendation", "test_hypothesis"],
        },
    },
}

variant_texts = "\n\n---\n\n".join(
    f"### {name}\n{data['output'][:2000]}"
    for name, data in variants.items()
)

compare_resp = mavera.responses.create(
    model="mavera-1",
    input=[
        {
            "role": "user",
            "content": (
                "Compare these A/B content variants. Score each 1-10 on emotional appeal, "
                "clarity, urgency, and CTA strength. Identify strengths and weaknesses. "
                "Recommend which to test first and state the hypothesis.\n\n"
                f"{variant_texts}"
            ),
        },
    ],
    extra_body={"response_format": COMPARISON_SCHEMA},
)

comparison = json.loads(compare_resp.output[0].content[0].text)
print(json.dumps(comparison, indent=2))