import os, requests, time
from collections import defaultdict
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"}
ORG_URN = "urn:li:organization:12345678"
# 1. Pull follower statistics
r = requests.get(f"{LI_BASE}/organizationalEntityFollowerStatistics",
headers=LI_H,
params={
"q": "organizationalEntity",
"organizationalEntity": ORG_URN,
})
if r.status_code == 429:
time.sleep(int(r.headers.get("Retry-After", 60)))
r = requests.get(f"{LI_BASE}/organizationalEntityFollowerStatistics",
headers=LI_H,
params={"q": "organizationalEntity", "organizationalEntity": ORG_URN})
r.raise_for_status()
stats = r.json().get("elements", [{}])[0]
# 2. Parse demographic breakdowns
def parse_breakdown(data, key):
result = {}
for entry in data:
name = entry.get(key, {}).get("localized", {}).get("en_US", "")
if not name:
name = entry.get(key, "Unknown")
organic = entry.get("followerCounts", {}).get("organicFollowerCount", 0)
paid = entry.get("followerCounts", {}).get("paidFollowerCount", 0)
result[name] = organic + paid
return dict(sorted(result.items(), key=lambda x: -x[1]))
industries = parse_breakdown(stats.get("followerCountsByIndustry", []), "industry")
functions = parse_breakdown(stats.get("followerCountsByFunction", []), "function")
seniorities = parse_breakdown(stats.get("followerCountsBySeniority", []), "seniority")
total = sum(industries.values()) or 1
print(f"Total followers: {total}")
print(f"Top industries: {list(industries.items())[:5]}")
print(f"Top functions: {list(functions.items())[:5]}")
print(f"Top seniorities: {list(seniorities.items())[:5]}")
# 3. Fetch existing Mavera personas
existing = requests.get(f"{MV_BASE}/personas", headers=MV_H).json()
li_personas = {p["name"]: p for p in (existing if isinstance(existing, list) else [])
if "LinkedIn Follower" in p.get("name", "")}
# 4. Build persona segments from top combinations
top_industries = list(industries.keys())[:4]
top_functions = list(functions.keys())[:3]
top_seniorities = list(seniorities.keys())[:3]
created, updated = [], []
for industry in top_industries:
ind_pct = round((industries[industry] / total) * 100, 1)
if ind_pct < 3:
continue
top_func = top_functions[0] if top_functions else "General"
top_sen = top_seniorities[0] if top_seniorities else "Senior"
name = f"LinkedIn Follower: {industry}"
desc = (
f"Derived from Company Page follower analytics. "
f"Industry: {industry} ({ind_pct}% of followers). "
f"Top function: {top_func} ({round((functions.get(top_func, 0) / total) * 100, 1)}%). "
f"Top seniority: {top_sen} ({round((seniorities.get(top_sen, 0) / total) * 100, 1)}%). "
f"Total follower base: {total:,}."
)
payload = {
"name": name,
"description": desc,
"demographic": {
"industries": [industry],
"job_titles": [top_func],
"seniority": top_sen,
},
"psychographic": {
"source": "linkedin_company_page_followers",
"audience_share_pct": ind_pct,
},
}
if name in li_personas:
r = requests.patch(f"{MV_BASE}/personas/{li_personas[name]['id']}",
headers=MV_H, json=payload)
r.raise_for_status()
updated.append({"name": name, "id": li_personas[name]["id"], "pct": ind_pct})
else:
r = requests.post(f"{MV_BASE}/personas", headers=MV_H, json=payload)
r.raise_for_status()
created.append({"name": name, "id": r.json()["id"], "pct": ind_pct})
time.sleep(0.3)
print(f"\nCreated {len(created)} | Updated {len(updated)} personas")
for p in created + updated:
action = "Created" if p in created else "Updated"
print(f" {action}: {p['name']} ({p['pct']}%) → {p['id']}")
