import os, requests, time, tempfile
META = os.environ["META_ACCESS_TOKEN"]
ACCT = os.environ["META_AD_ACCOUNT_ID"]
MV = os.environ["MAVERA_API_KEY"]
GRAPH = "https://graph.facebook.com/v24.0"
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
# 1. Pull active ads with creative details
ads = requests.get(
f"{GRAPH}/{ACCT}/ads",
params={
"access_token": META,
"effective_status": '["ACTIVE"]',
"fields": "id,name,creative{id,name,video_id,title,body}",
"limit": 50,
},
).json().get("data", [])
video_ads = [a for a in ads if a.get("creative", {}).get("video_id")]
print(f"Active video ads: {len(video_ads)}")
# 2. Analyze each (reusing upload+analysis pattern)
scored = []
for ad in video_ads[:15]:
creative = ad["creative"]
vid = creative["video_id"]
video_info = requests.get(
f"{GRAPH}/{vid}",
params={"access_token": META, "fields": "source,length"},
).json()
if not video_info.get("source"):
continue
vid_resp = requests.get(video_info["source"], stream=True)
tmp = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
for chunk in vid_resp.iter_content(8192):
tmp.write(chunk)
tmp.close()
with open(tmp.name, "rb") as f:
asset = requests.post(f"{MB}/assets",
headers={"Authorization": f"Bearer {MV}"},
files={"file": (f"{vid}.mp4", f, "video/mp4")},
).json()
os.unlink(tmp.name)
analysis = requests.post(f"{MB}/video-analyses", headers=MH,
json={"asset_id": asset["id"], "name": creative.get("name", vid)},
).json()
for _ in range(30):
time.sleep(10)
status = requests.get(f"{MB}/video-analyses/{analysis['id']}",
headers={"Authorization": f"Bearer {MV}"}).json()
if status.get("status") in ("completed", "failed"):
break
if status.get("status") == "completed":
scored.append({
"ad_name": ad.get("name", "Untitled"),
"creative_name": creative.get("name", ""),
"title": creative.get("title", ""),
"body": creative.get("body", "")[:100],
"duration": video_info.get("length"),
"scores": status.get("scores", {}),
})
time.sleep(1)
# 3. Build comparison table for Mave
table_rows = []
for i, s in enumerate(scored, 1):
sc = s["scores"]
table_rows.append(
f"{i}. \"{s['ad_name']}\" — emotional: {sc.get('emotional','?')}, "
f"cognitive: {sc.get('cognitive','?')}, behavioral: {sc.get('behavioral','?')}, "
f"duration: {s['duration']}s, copy: \"{s['body']}\""
)
table = "\n".join(table_rows)
# 4. Mave comparison
comparison = requests.post(f"{MB}/mave/chat", headers=MH, json={
"message": f"""Compare these {len(scored)} video ad creatives based on their Mavera analysis scores.
CREATIVE SCORES:
{table}
Produce:
1. Ranked matrix (best to worst) with rationale
2. Message clarity ranking
3. Emotional intensity ranking
4. CTA strength ranking
5. Which creatives to SCALE (top 3) and why
6. Which creatives to PAUSE or REWORK and what to fix
7. Patterns in top performers vs bottom performers
8. Specific recommendations for each creative"""
}).json()
print("=== Creative Comparison Matrix ===")
print(comparison.get("content", ""))