import os, requests, time
from google.ads.googleads.client import GoogleAdsClient
MV = os.environ["MAVERA_API_KEY"]
CUSTOMER_ID = os.environ["GOOGLE_ADS_CUSTOMER_ID"]
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
client = GoogleAdsClient.load_from_env()
ga_service = client.get_service("GoogleAdsService")
query = """
SELECT
ad_group_ad.ad.id,
ad_group_ad.ad.responsive_search_ad.headlines,
ad_group_ad.ad.responsive_search_ad.descriptions,
metrics.impressions,
metrics.conversions,
metrics.ctr
FROM ad_group_ad
WHERE ad_group_ad.ad.type = RESPONSIVE_SEARCH_AD
AND segments.date DURING LAST_30_DAYS
AND metrics.impressions > 500
ORDER BY metrics.impressions DESC
LIMIT 5
"""
response = ga_service.search(customer_id=CUSTOMER_ID, query=query)
combos = []
for row in response:
rsa = row.ad_group_ad.ad.responsive_search_ad
best_headlines = [
a.text for a in rsa.headlines
if a.performance_label and a.performance_label.name in ("BEST", "GOOD")
][:3] or [a.text for a in rsa.headlines[:3]]
best_descs = [
a.text for a in rsa.descriptions
if a.performance_label and a.performance_label.name in ("BEST", "GOOD")
][:2] or [a.text for a in rsa.descriptions[:2]]
combos.append({
"ad_id": row.ad_group_ad.ad.id,
"headlines": best_headlines,
"descriptions": best_descs,
"impressions": row.metrics.impressions,
"ctr": row.metrics.ctr,
"conversions": row.metrics.conversions,
"all_headlines": [{"text": a.text, "perf": a.performance_label.name if a.performance_label else "UNKNOWN"} for a in rsa.headlines],
})
PERSONA_IDS = os.environ.get("RSA_PERSONA_IDS", "").split(",")
if not PERSONA_IDS[0]:
for name, desc in [
("Startup Founder", "CEO/Founder of a 10-50 person startup. Time-poor, skeptical of marketing claims."),
("Marketing Director", "Director of Marketing at mid-market B2B. Manages team of 5-10. Accountable for pipeline."),
("Procurement Manager", "Procurement lead at enterprise. Risk-averse, process-oriented, price-sensitive."),
]:
p = requests.post(f"{MB}/personas", headers=MH, json={"name": name, "description": desc}).json()
PERSONA_IDS.append(p["id"])
time.sleep(0.3)
combo_block = "\n\n".join(
f"Combo {i+1} (CTR: {c['ctr']:.2%}, Conv: {c['conversions']:.0f}):\n"
f" Headlines: {' | '.join(c['headlines'])}\n"
f" Descriptions: {' | '.join(c['descriptions'])}"
for i, c in enumerate(combos)
)
fg = requests.post(f"{MB}/focus-groups", headers=MH, json={
"name": "RSA Headline/Description Testing",
"persona_ids": [pid for pid in PERSONA_IDS if pid],
"questions": [
f"Here are 5 ad combinations:\n\n{combo_block}\n\nRank them 1-5 and explain your top pick.",
"Which headline would stop your scroll? What makes it stand out?",
"Which description would convince you to click? Why?",
"Rewrite the weakest headline to make it compelling for someone like you.",
],
"context": "These are Google Search ads for a B2B marketing platform.",
"responses_per_persona": 2,
}).json()
for _ in range(20):
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')}")
for resp in data.get("responses", []):
print(f"\n[{resp.get('persona_id','?')}] {resp.get('question','')[:60]}")
print(f" → {resp.get('answer','')[:300]}")
print("\n--- Asset Performance Summary ---")
for c in combos:
for h in c["all_headlines"]:
print(f" [{h['perf']:8s}] {h['text']}")
