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 all creatives
creatives_resp = requests.get(f"{LI_BASE}/adAccounts/{AD_ACCOUNT_ID}/creatives",
headers=LI_H,
params={"q": "search", "count": 50})
if creatives_resp.status_code == 429:
time.sleep(int(creatives_resp.headers.get("Retry-After", 60)))
creatives_resp = requests.get(f"{LI_BASE}/adAccounts/{AD_ACCOUNT_ID}/creatives",
headers=LI_H, params={"q": "search", "count": 50})
creatives_resp.raise_for_status()
creatives = creatives_resp.json().get("elements", [])
# 2. Pull analytics for each creative
creative_data = []
for cr in creatives:
cr_urn = f"urn:li:sponsoredCreative:{cr.get('id', '')}"
r = requests.get(f"{LI_BASE}/adAnalytics",
headers=LI_H,
params={
"q": "analytics", "pivot": "CREATIVE",
"timeGranularity": "ALL",
"creatives[0]": cr_urn,
"dateRange.start.year": 2025, "dateRange.start.month": 1, "dateRange.start.day": 1,
"dateRange.end.year": 2025, "dateRange.end.month": 12, "dateRange.end.day": 31,
"fields": "impressions,clicks,likes,comments,shares",
})
if r.status_code == 429:
time.sleep(int(r.headers.get("Retry-After", 60)))
continue
metrics = r.json().get("elements", [{}])[0] if r.ok else {}
copy = cr.get("commentary", "") or cr.get("content", {}).get("textAd", {}).get("text", "")
headline = cr.get("content", {}).get("textAd", {}).get("headline", "")
impressions = metrics.get("impressions", 0)
clicks = metrics.get("clicks", 0)
if copy and impressions > 0:
creative_data.append({
"copy": copy, "headline": headline,
"impressions": impressions, "clicks": clicks,
"ctr": round((clicks / impressions) * 100, 2) if impressions else 0,
"engagement": metrics.get("likes", 0) + metrics.get("comments", 0) + metrics.get("shares", 0),
})
time.sleep(0.3)
# 3. Create Brand Voice from top performers
top_ads = sorted(creative_data, key=lambda x: -x["ctr"])[:10]
samples = "\n\n---\n\n".join(
f"Headline: {a['headline']}\nCopy: {a['copy'][:500]}\nCTR: {a['ctr']}%"
for a in top_ads
)
bv = requests.post(f"{MV_BASE}/brand-voices", headers=MV_H, json={
"name": "LinkedIn Top-Performing Ad Voice",
"samples": [samples],
}).json()
print(f"Brand Voice: {bv['id']}")
# 4. Generate new ad copy variants
gen = requests.post(f"{MV_BASE}/generations", headers=MV_H, json={
"brand_voice_id": bv["id"],
"app_template": "linkedin_ad",
"prompt": (
"Generate 5 LinkedIn Sponsored Content ad variants for a B2B SaaS product. "
"Each variant needs: intro text (under 150 chars for mobile), headline, "
"description, and CTA button text. Match the winning voice from our top ads. "
"Vary the hooks: use a stat, a question, a bold claim, social proof, and a trend."
),
"count": 5,
}).json()
for i, g in enumerate(gen.get("results", [gen]), 1):
print(f"\n--- Variant {i} ---")
print(g.get("content", g.get("text", ""))[:400])
# 5. Performance benchmark
avg_ctr = sum(a["ctr"] for a in top_ads) / len(top_ads) if top_ads else 0
print(f"\nBenchmark: Top-performer avg CTR = {avg_ctr:.2f}%")
print(f"Generated {len(gen.get('results', [gen]))} variants with brand voice {bv['id']}")
