#!/usr/bin/env python3
"""
02_ad_copy_performance_analyzer.py — Analyze ad copy (headlines, descriptions,
CTAs) across Google Ads & Meta Ads to find winning messaging patterns
via GoMarble MCP tools + Claude API.
Outputs: ad_copy_performance.csv, ad_copy_brief.txt
"""
# ┌──────────────────────────────────────────────────────────┐
# │ EDIT THESE VALUES BEFORE RUNNING │
# └──────────────────────────────────────────────────────────┘
ANTHROPIC_API_KEY = "" # Your Anthropic API key
GOMARBLE_API_KEY = "" # From GoMarble dashboard
META_AD_ACCOUNT = "" # e.g. "act_123456" (leave "" to skip Meta)
GOOGLE_ADS_CID = "" # e.g. "7556164258" (leave "" to skip Google)
# NOTE: At least one of META_AD_ACCOUNT or GOOGLE_ADS_CID must be set.
import csv, json, re, sys, time, subprocess
try:
import requests
except ImportError:
subprocess.check_call([sys.executable, "-m", "pip", "install", "requests"])
import requests
API_URL = "https://api.anthropic.com/v1/messages"
MODEL = "claude-sonnet-4-20250514"
def safe_float(val, default=0.0):
if val is None: return default
try:
s = str(val).replace(",", "").replace("$", "").replace("%", "").strip()
return float(s) if s else default
except (ValueError, TypeError): return default
def try_parse_json(text):
text = re.sub(r"```(?:json)?\s*", "", text)
for open_ch, close_ch in [("{", "}"), ("[", "]")]:
pos = 0
while pos < len(text):
start = text.find(open_ch, pos)
if start == -1: break
depth = 0
for i in range(start, len(text)):
if text[i] == open_ch: depth += 1
elif text[i] == close_ch: depth -= 1
if depth == 0:
try: return json.loads(text[start : i + 1])
except json.JSONDecodeError: pass
break
pos = start + 1
return None
def to_rows(text):
data = try_parse_json(text)
if data is None: return []
if isinstance(data, list): return data
if isinstance(data, dict):
for key in ("results", "rows", "data", "ads", "creatives"):
if key in data and isinstance(data[key], list): return data[key]
return [data]
return []
def flatten_row(row):
if not isinstance(row, dict): return {}
flat = {}
def _flatten(obj, prefix=""):
if isinstance(obj, dict):
for k, v in obj.items(): _flatten(v, f"{prefix}_{k}" if prefix else k)
elif isinstance(obj, list):
# Join list items (e.g. headlines, descriptions, final_urls)
str_items = []
for item in obj:
if isinstance(item, dict):
# Handle Google Ads asset format: {"text": "...", "pinnedField": ...}
txt = item.get("text") or item.get("value") or item.get("name")
if txt: str_items.append(str(txt))
else: str_items.append(json.dumps(item))
else:
str_items.append(str(item))
flat[prefix] = " | ".join(str_items) if str_items else ""
else:
flat[prefix] = obj
_flatten(row)
return flat
def normalize_google_row(flat):
"""Map Google Ads flattened keys to standard field names."""
mapping = {
"campaign_name": ["campaign_name"],
"ad_name": ["ad_group_ad_ad_name", "adGroupAd_ad_name", "ad_name"],
"ad_type": ["ad_group_ad_ad_type", "adGroupAd_ad_type", "ad_type"],
"headlines": ["ad_group_ad_ad_responsive_search_ad_headlines",
"adGroupAd_ad_responsiveSearchAd_headlines",
"responsive_search_ad_headlines", "headlines"],
"descriptions": ["ad_group_ad_ad_responsive_search_ad_descriptions",
"adGroupAd_ad_responsiveSearchAd_descriptions",
"responsive_search_ad_descriptions", "descriptions"],
"final_urls": ["ad_group_ad_ad_final_urls", "adGroupAd_ad_finalUrls", "final_urls"],
"impressions": ["metrics_impressions"],
"clicks": ["metrics_clicks"],
"cost": ["metrics_cost_micros"],
"conversions": ["metrics_conversions"],
"conversions_value": ["metrics_conversions_value", "metrics_conversionsValue"],
"ctr": ["metrics_ctr"],
"cpc": ["metrics_average_cpc", "metrics_averageCpc"],
}
norm = {}
for target, sources in mapping.items():
for src in sources:
if src in flat and flat[src] not in (None, "", "0"):
norm[target] = flat[src]
break
if target not in norm:
if target in flat:
norm[target] = flat[target]
# Convert cost_micros to dollars
if "cost" in norm:
try:
norm["cost"] = round(safe_float(norm["cost"]) / 1_000_000, 2)
except Exception:
pass
for k, v in flat.items():
if k not in norm and v not in (None, ""):
norm[k] = v
return norm
def normalize_meta_row(flat):
"""Map Meta Ads flattened keys to standard field names."""
mapping = {
"ad_name": ["ad_name", "name"],
"campaign_name": ["campaign_name"],
"spend": ["spend"],
"impressions": ["impressions"],
"clicks": ["clicks"],
"ctr": ["ctr"],
"cpc": ["cpc"],
"purchase_roas": ["purchase_roas", "roas"],
"conversions": ["conversions", "actions_purchase"],
}
norm = {}
for target, sources in mapping.items():
for src in sources:
if src in flat and flat[src] not in (None, "", "0"):
norm[target] = flat[src]
break
if target not in norm and target in flat:
norm[target] = flat[target]
for k, v in flat.items():
if k not in norm and v not in (None, ""):
norm[k] = v
return norm
def banner(step, total, title):
print(f"\n{'─'*60}\n Step {step}/{total} │ {title}\n{'─'*60}")
def write_csv(path, rows, fields):
if not rows: print(f" (no data for {path})"); return
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
w.writeheader(); w.writerows(rows)
print(f" -> {path} — {len(rows)} rows")
def mcp_request(messages, system=None, max_tokens=16000):
headers = {"x-api-key": ANTHROPIC_API_KEY, "anthropic-version": "2023-06-01",
"anthropic-beta": "mcp-client-2025-04-04", "content-type": "application/json"}
payload = {"model": MODEL, "max_tokens": max_tokens, "messages": messages,
"mcp_servers": [{"type": "url", "url": "https://apps.gomarble.ai/mcp-api/sse",
"name": "gomarble", "authorization_token": GOMARBLE_API_KEY}]}
if system: payload["system"] = system
for attempt in range(3):
try:
resp = requests.post(API_URL, headers=headers, json=payload, timeout=300)
if resp.status_code != 200: print(f" API error {resp.status_code}: {resp.text[:300]}"); resp.raise_for_status()
return resp.json()
except requests.exceptions.ReadTimeout:
print(f" Timeout (attempt {attempt+1}/3), retrying...")
if attempt == 2: raise
time.sleep(5)
except requests.exceptions.ConnectionError:
print(f" Connection error (attempt {attempt+1}/3), retrying..."); time.sleep(5)
if attempt == 2: raise
def get_text(response):
parts = []
for block in response.get("content", []):
if block.get("type") == "text": parts.append(block["text"])
elif block.get("type") == "mcp_tool_result":
content = block.get("content", "")
if isinstance(content, list):
for c in content:
if isinstance(c, dict) and c.get("text"): parts.append(c["text"])
elif isinstance(content, str): parts.append(content)
return "\n".join(parts)
# ── Step 1: Meta Ad Copy + Performance ────────────────────────
def step1_meta_ad_copy(meta_id):
banner(1, 5, "Meta Ad Copy & Performance (30d)")
if not meta_id: print(" Skipped (no Meta account)"); return None, None
print(" Fetching ad-level performance...")
msg = (
f"Run facebook_get_adaccount_insights with ad_account_id='{meta_id}', "
"level='ad', date_preset='last_30d', "
"fields=['ad_name','campaign_name','spend','impressions','clicks','ctr',"
"'cpc','purchase_roas','actions']. "
"Return the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
perf_text = get_text(resp)
with open("debug_meta_perf_raw.txt", "w", encoding="utf-8") as f:
f.write(perf_text)
print(f" Ads: {len(to_rows(perf_text))} (saved to debug_meta_perf_raw.txt)")
print(" Fetching creative details (copy, headlines, CTAs)...")
msg2 = (
f"Run facebook_get_ad_creative_details with ad_account_id='{meta_id}'. "
"Return the raw JSON with body text, headlines, descriptions, "
"call_to_action_type, link_url for each ad creative."
)
resp2 = mcp_request([{"role": "user", "content": msg2}])
creative_text = get_text(resp2)
with open("debug_meta_creative_raw.txt", "w", encoding="utf-8") as f:
f.write(creative_text)
print(f" Creative details: {len(creative_text)} chars (saved to debug_meta_creative_raw.txt)")
return perf_text, creative_text
# ── Step 2: Google Ad Copy + Performance ──────────────────────
def step2_google_ad_copy(google_id):
banner(2, 5, "Google Ad Copy & Performance (30d)")
if not google_id: print(" Skipped (no Google account)"); return None
msg = (
f"Run google_ads_run_gaql with customer_id='{google_id}' and query:\n"
"SELECT ad_group_ad.ad.name, ad_group_ad.ad.type, "
"ad_group_ad.ad.responsive_search_ad.headlines, "
"ad_group_ad.ad.responsive_search_ad.descriptions, "
"ad_group_ad.ad.final_urls, campaign.name, "
"metrics.impressions, metrics.clicks, metrics.cost_micros, "
"metrics.conversions, metrics.conversions_value, metrics.ctr, "
"metrics.average_cpc "
"FROM ad_group_ad WHERE segments.date DURING LAST_30_DAYS "
"AND campaign.status = 'ENABLED' AND ad_group_ad.status = 'ENABLED' "
"ORDER BY metrics.impressions DESC LIMIT 100\n\n"
"Return the COMPLETE raw JSON result — do NOT summarize or truncate. "
"Output the full JSON array of all rows."
)
resp = mcp_request([{"role": "user", "content": msg}])
text = get_text(resp)
with open("debug_google_raw.txt", "w", encoding="utf-8") as f:
f.write(text)
print(f" Raw response: {len(text)} chars (saved to debug_google_raw.txt)")
rows = to_rows(text)
print(f" Google ads parsed: {len(rows)} rows")
if rows:
print(f" Sample row keys: {list(rows[0].keys())[:15]}")
return text
# ── Step 3: Generate Copy Analysis ────────────────────────────
def step3_generate_analysis(meta_perf, meta_creative, google_text):
banner(3, 5, "AI Copy Analysis")
prompt = f"""You are a senior copywriter & performance marketer. Analyze ad copy across platforms.
Sections:
1. COPY PERFORMANCE OVERVIEW — Total ads analyzed, average CTR/CPC by copy style
2. TOP PERFORMING COPY — Top 10 ads by ROAS/CTR with their exact headlines & body text.
For each, explain WHY the copy works (emotional triggers, urgency, specificity, etc.)
3. UNDERPERFORMING COPY — Bottom 10 ads with their copy. Diagnose specific copy issues:
- Weak headlines (no benefit, no urgency)
- Generic body text
- Mismatched CTA
- Missing social proof
4. HEADLINE ANALYSIS — Which headline patterns drive highest CTR:
- Question vs statement
- Number-driven vs emotional
- Benefit-first vs feature-first
- Length analysis (short vs long)
5. CTA ANALYSIS — Which call-to-action types perform best (Shop Now, Learn More, etc.)
6. MESSAGING THEMES — Cluster ads by messaging angle (price, quality, urgency, social proof,
testimonial, benefit-driven) and compare performance
7. GOOGLE RSA INSIGHTS — (If available) Which headline/description combinations win
8. A/B TEST IDEAS — 5 specific copy tests to run with hypotheses
9. COPY PLAYBOOK — Template formulas for writing winning ads based on data:
- Headline formula
- Body copy formula
- CTA pairing recommendations
Use exact ad copy text and metrics throughout.
### Meta Ad Performance
{meta_perf[:8000] if meta_perf else 'N/A'}
### Meta Creative Details (copy, headlines, CTAs)
{meta_creative[:8000] if meta_creative else 'N/A'}
### Google Ad Copy & Performance
{google_text[:12000] if google_text else 'N/A'}"""
resp = mcp_request([{"role": "user", "content": prompt}])
analysis = get_text(resp)
print(f" Analysis: {len(analysis)} chars")
return analysis
# ── Step 4: Build CSV Data ────────────────────────────────────
def step4_build_csv(meta_perf, google_text):
banner(4, 5, "Build CSV Data")
rows = []
if meta_perf:
for row in to_rows(meta_perf):
flat = flatten_row(row)
norm = normalize_meta_row(flat)
norm["platform"] = "Meta"
rows.append(norm)
if google_text:
for row in to_rows(google_text):
flat = flatten_row(row)
norm = normalize_google_row(flat)
norm["platform"] = "Google"
rows.append(norm)
print(f" Total rows: {len(rows)}")
if rows:
print(f" Sample normalized keys: {list(rows[0].keys())[:15]}")
return rows
# ── Step 5: Output ────────────────────────────────────────────
def step5_output(rows, analysis):
banner(5, 5, "Write Output Files")
if rows:
preferred = ["platform", "campaign_name", "ad_name", "ad_type",
"headlines", "descriptions", "final_urls",
"impressions", "clicks", "ctr", "cpc", "cost",
"conversions", "conversions_value", "purchase_roas",
"spend", "body", "headline", "description",
"call_to_action_type", "link_url"]
all_keys = []
seen = set()
for r in rows:
for k in r:
if k not in seen:
seen.add(k)
all_keys.append(k)
fields = [f for f in preferred if f in seen]
fields += [k for k in all_keys if k not in fields]
write_csv("ad_copy_performance.csv", rows, fields)
with open("ad_copy_brief.txt", "w", encoding="utf-8") as f:
f.write("AD COPY PERFORMANCE ANALYSIS\n")
f.write(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write("=" * 60 + "\n\n")
f.write(analysis)
print(f" -> ad_copy_brief.txt — {len(analysis)} chars")
print(f"\n{'='*60}\n AD COPY ANALYSIS COMPLETE\n{'='*60}")
def main():
meta_id = META_AD_ACCOUNT.strip() or None
google_id = GOOGLE_ADS_CID.strip().replace("-", "") or None
if not meta_id and not google_id:
sys.exit("Set at least one of META_AD_ACCOUNT or GOOGLE_ADS_CID at the top of the file.")
if "..." in ANTHROPIC_API_KEY or "xxxx" in GOMARBLE_API_KEY:
sys.exit("Edit ANTHROPIC_API_KEY and GOMARBLE_API_KEY at the top of the file.")
print(f"\nAd Copy Performance Analyzer")
if meta_id: print(f" Meta: {meta_id}")
if google_id: print(f" Google: {google_id}")
meta_perf, meta_creative = step1_meta_ad_copy(meta_id) if meta_id else (None, None)
google_text = step2_google_ad_copy(google_id)
analysis = step3_generate_analysis(meta_perf, meta_creative, google_text)
rows = step4_build_csv(meta_perf, google_text)
step5_output(rows, analysis)
if __name__ == "__main__":
main()