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 performance by placement for recent ads
placement_data = requests.get(
f"{GRAPH}/{ACCT}/insights",
params={
"access_token": META,
"fields": "ad_id,ad_name,impressions,clicks,ctr,cpc,spend,actions",
"breakdowns": "publisher_platform,platform_position",
"level": "ad",
"date_preset": "last_30d",
"limit": 200,
},
).json().get("data", [])
# 2. Group by ad, find ads with multi-placement data
from collections import defaultdict
ad_placements = defaultdict(list)
for row in placement_data:
ad_placements[row["ad_id"]].append({
"platform": row.get("publisher_platform", "unknown"),
"position": row.get("platform_position", "unknown"),
"impressions": int(row.get("impressions", 0)),
"clicks": int(row.get("clicks", 0)),
"ctr": float(row.get("ctr", 0)),
"spend": float(row.get("spend", 0)),
"name": row.get("ad_name", ""),
})
multi_placement = {k: v for k, v in ad_placements.items() if len(v) >= 2}
print(f"Ads with 2+ placements: {len(multi_placement)}")
# 3. Pick the top ad by total spend and analyze
if not multi_placement:
print("No multi-placement ads found.")
exit()
target_ad_id = max(multi_placement, key=lambda k: sum(p["spend"] for p in multi_placement[k]))
placements = multi_placement[target_ad_id]
ad_name = placements[0]["name"]
print(f"Analyzing: {ad_name} across {len(placements)} placements")
# 4. Get the creative video
ad_detail = requests.get(
f"{GRAPH}/{target_ad_id}",
params={"access_token": META, "fields": "creative{video_id,title,body}"},
).json()
vid = ad_detail.get("creative", {}).get("video_id")
if not vid:
print("Selected ad has no video creative.")
exit()
video_info = requests.get(f"{GRAPH}/{vid}",
params={"access_token": META, "fields": "source,length"}).json()
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()
# 5. Run Video Analysis for each placement context
placement_analyses = []
for pl in placements:
label = f"{pl['platform']} — {pl['position']}"
with open(tmp.name, "rb") as f:
asset = requests.post(f"{MB}/assets",
headers={"Authorization": f"Bearer {MV}"},
files={"file": (f"{vid}_{pl['platform']}_{pl['position']}.mp4", f, "video/mp4")}).json()
analysis = requests.post(f"{MB}/video-analyses", headers=MH, json={
"asset_id": asset["id"],
"name": f"{ad_name} — {label}",
}).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":
placement_analyses.append({
"placement": label,
"platform": pl["platform"],
"position": pl["position"],
"ctr": pl["ctr"],
"spend": pl["spend"],
"scores": status.get("scores", {}),
})
time.sleep(1)
os.unlink(tmp.name)
# 6. Mave comparison
analysis_summary = "\n".join(
f"- {pa['placement']}: CTR={pa['ctr']:.2f}%, spend=${pa['spend']:.0f}, "
f"emotional={pa['scores'].get('emotional','?')}, cognitive={pa['scores'].get('cognitive','?')}, "
f"behavioral={pa['scores'].get('behavioral','?')}"
for pa in placement_analyses
)
comparison = requests.post(f"{MB}/mave/chat", headers=MH, json={
"message": f"""Compare this video creative's performance across Meta placements.
CREATIVE: "{ad_name}" ({video_info.get('length',0)}s video)
PLACEMENT ANALYSIS:
{analysis_summary}
For each placement:
1. Why does performance differ? (viewing context, user intent, format fit)
2. Specific edits for this placement (aspect ratio, hook timing, CTA overlay position)
3. Budget reallocation recommendation based on efficiency
4. Which placement should get the most budget and why
5. Should any placement get a dedicated creative variant?"""
}).json()
print("\n=== Cross-Platform Creative Analysis ===")
print(comparison.get("content", ""))
