import os, requests, time
LI = os.environ["LINKEDIN_ACCESS_TOKEN"]
MV = os.environ["MAVERA_API_KEY"]
LI_BASE = "https://api.linkedin.com/rest"
MV_BASE = "https://app.mavera.io/api/v1"
LI_H = {"Authorization": f"Bearer {LI}", "LinkedIn-Version": "202401", "X-Restli-Protocol-Version": "2.0.0"}
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
AD_ACCOUNT_ID = "508000001"
# 1. Pull active creatives
r = requests.get(f"{LI_BASE}/adAccounts/{AD_ACCOUNT_ID}/creatives",
headers=LI_H,
params={"q": "search", "search.status.values[0]": "ACTIVE", "count": 20})
if r.status_code == 429:
retry_after = int(r.headers.get("Retry-After", 60))
time.sleep(retry_after)
r = requests.get(f"{LI_BASE}/adAccounts/{AD_ACCOUNT_ID}/creatives",
headers=LI_H,
params={"q": "search", "search.status.values[0]": "ACTIVE", "count": 20})
r.raise_for_status()
creatives = r.json().get("elements", [])
# 2. Extract ad content
ads = []
for cr in creatives:
content = cr.get("content", {})
text = content.get("textAd", {}).get("text", "")
headline = content.get("textAd", {}).get("headline", "")
intro = cr.get("intendedStatus", "")
commentary = cr.get("commentary", "")
if commentary or text:
ads.append({
"id": cr.get("id", ""),
"copy": commentary or text,
"headline": headline,
"format": cr.get("content", {}).get("contentType", "SINGLE_IMAGE"),
})
if not ads:
raise SystemExit("No active creatives found. Check ad account ID and token scopes.")
# 3. Create B2B personas
ROLES = [
{"title": "VP of Marketing", "desc": "Senior marketer evaluating MarTech. Budget authority. Cares about ROI and team efficiency."},
{"title": "Director of Sales", "desc": "Sales leader. Evaluates tools for pipeline acceleration. Skeptical of marketing fluff."},
{"title": "Product Manager", "desc": "Builds product roadmaps. Evaluates solutions for user adoption and feature alignment."},
{"title": "CFO / Finance Lead", "desc": "Controls budget. Needs clear ROI justification. Risk-averse to new vendors."},
]
persona_ids = []
for role in ROLES:
p = requests.post(f"{MV_BASE}/personas", headers=MV_H, json={
"name": f"LinkedIn B2B: {role['title']}",
"description": role["desc"],
"demographic": {"job_titles": [role["title"]]},
}).json()
persona_ids.append({"id": p["id"], "title": role["title"]})
time.sleep(0.3)
# 4. Build stimulus from ads
stimulus = "\n\n---\n\n".join(
f"AD {i+1} ({a['format']}):\nHeadline: {a['headline']}\nCopy: {a['copy'][:400]}"
for i, a in enumerate(ads[:5])
)
# 5. Run Focus Group
fg = requests.post(f"{MV_BASE}/focus-groups", headers=MV_H, json={
"name": "LinkedIn Sponsored Content Review",
"persona_ids": [p["id"] for p in persona_ids],
"questions": [
f"Review these LinkedIn ads:\n\n{stimulus}\n\nRate each ad 1-5 for relevance to your role (1=irrelevant, 5=highly relevant). Explain your ratings.",
"Which ad would you most likely click on in your LinkedIn feed? Why?",
"What is missing from these ads that would make them more compelling for someone in your position?",
"If you saw this ad from a competitor, would it make you reconsider your current solution?",
],
"responses_per_persona": 2,
}).json()
# 6. Poll for results
for _ in range(20):
time.sleep(5)
data = requests.get(f"{MV_BASE}/focus-groups/{fg['id']}", headers=MV_H).json()
if data.get("status") == "completed":
break
for resp in data.get("responses", []):
title = next((p["title"] for p in persona_ids if p["id"] == resp.get("persona_id")), "?")
print(f"[{title}] {resp.get('question','')[:60]}...")
print(f" → {resp.get('answer','')[:300]}\n")