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 Instagram placement ads (Reels)
ads = requests.get(
f"{GRAPH}/{ACCT}/ads",
params={
"access_token": META,
"fields": "id,name,creative{id,name,video_id,title,body,instagram_permalink_url},targeting{publisher_platforms}",
"effective_status": '["ACTIVE","PAUSED"]',
"limit": 50,
},
).json().get("data", [])
reels = [a for a in ads if a.get("creative", {}).get("video_id")
and "instagram" in str(a.get("targeting", {}).get("publisher_platforms", [])).lower()]
print(f"Instagram Reels creatives: {len(reels)}")
# 2. Video Analysis on each
analyses = []
for reel in reels[:8]:
vid = reel["creative"]["video_id"]
video_info = requests.get(f"{GRAPH}/{vid}",
params={"access_token": META, "fields": "source,length,title"}).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"reel_{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": f"Reel: {reel.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":
analyses.append({
"name": reel.get("name", vid),
"video_id": vid,
"duration": video_info.get("length"),
"body": reel["creative"].get("body", "")[:150],
"scores": status.get("scores", {}),
"analysis_id": analysis["id"],
})
time.sleep(1)
# 3. Focus Group on hook, recall, purchase intent
reel_descriptions = "\n\n".join(
f"Reel {i+1}: \"{a['name']}\" ({a['duration']}s)\n"
f"Copy: {a['body']}\n"
f"Emotional score: {a['scores'].get('emotional','?')}, "
f"Hook strength: {a['scores'].get('attention_curve',['?'])[0] if isinstance(a['scores'].get('attention_curve'), list) else '?'}"
for i, a in enumerate(analyses)
)
PERSONA_IDS = os.environ.get("PERSONA_IDS", "").split(",")
if not PERSONA_IDS[0]:
personas = requests.get(f"{MB}/personas",
headers={"Authorization": f"Bearer {MV}"}).json()
PERSONA_IDS = [p["id"] for p in (personas if isinstance(personas, list) else [])[:4]]
fg = requests.post(f"{MB}/focus-groups", headers=MH, json={
"name": "Instagram Reels Analysis",
"persona_ids": PERSONA_IDS,
"questions": [
f"Based on these Reels descriptions, which hook would stop your scroll? Why?\n\n{reel_descriptions}",
"Rate your brand recall after seeing each Reel (1-10). Which brand/product sticks?",
"Which Reel would make you most likely to purchase or visit a website? Why?",
"What makes a Reel hook effective for you? First 3 seconds — what grabs attention?",
"Would you share any of these Reels? Which one and why?",
],
"context": "You are evaluating Instagram Reels ads for effectiveness.",
"responses_per_persona": 2,
}).json()
for _ in range(24):
time.sleep(5)
fg_data = requests.get(f"{MB}/focus-groups/{fg['id']}", headers=MH).json()
if fg_data.get("status") == "completed":
break
# 4. Combined report
print("=== Instagram Reels Analysis Report ===\n")
print("--- Video Analysis Scores ---")
for a in analyses:
s = a["scores"]
print(f" {a['name']}: emotional={s.get('emotional','?')} cognitive={s.get('cognitive','?')} behavioral={s.get('behavioral','?')}")
print("\n--- Focus Group Insights ---")
for resp in fg_data.get("responses", [])[:12]:
print(f" [{resp.get('persona_id','?')[:12]}] {resp.get('question','')[:50]}...")
print(f" → {resp.get('answer','')[:200]}\n")
