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"}
ORG_URN = "urn:li:organization:12345678"
# 1. Pull recent company posts
r = requests.get(f"{LI_BASE}/posts",
headers=LI_H,
params={
"q": "author",
"author": ORG_URN,
"count": 50,
"sortBy": "LAST_MODIFIED",
})
if r.status_code == 429:
time.sleep(int(r.headers.get("Retry-After", 60)))
r = requests.get(f"{LI_BASE}/posts", headers=LI_H,
params={"q": "author", "author": ORG_URN, "count": 50, "sortBy": "LAST_MODIFIED"})
r.raise_for_status()
posts = r.json().get("elements", [])
# 2. Fetch engagement for each post
enriched = []
for post in posts:
post_urn = post.get("id", "")
commentary = post.get("commentary", "")
if not commentary:
continue
sr = requests.get(f"{LI_BASE}/socialActions/{post_urn}",
headers=LI_H)
if sr.status_code == 429:
time.sleep(int(sr.headers.get("Retry-After", 30)))
sr = requests.get(f"{LI_BASE}/socialActions/{post_urn}", headers=LI_H)
if sr.ok:
actions = sr.json()
likes = actions.get("likesSummary", {}).get("totalLikes", 0)
comments = actions.get("commentsSummary", {}).get("totalFirstLevelComments", 0)
shares = actions.get("sharesSummary", {}).get("totalShares", 0)
else:
likes = comments = shares = 0
total_engagement = likes + comments * 3 + shares * 5
enriched.append({
"urn": post_urn,
"commentary": commentary[:500],
"content_type": post.get("content", {}).get("contentType", "NONE"),
"likes": likes,
"comments": comments,
"shares": shares,
"engagement_score": total_engagement,
})
time.sleep(0.4)
# 3. Rank and take top performers
ranked = sorted(enriched, key=lambda x: -x["engagement_score"])
top = ranked[:10]
bottom = ranked[-5:] if len(ranked) >= 15 else []
# 4. Build analysis prompt
top_block = "\n\n---\n\n".join(
f"POST {i+1} (engagement: {p['engagement_score']} | {p['likes']} likes, {p['comments']} comments, {p['shares']} shares | {p['content_type']}):\n{p['commentary']}"
for i, p in enumerate(top)
)
bottom_block = "\n\n---\n\n".join(
f"LOW {i+1} (engagement: {p['engagement_score']}):\n{p['commentary']}"
for i, p in enumerate(bottom)
) if bottom else "No low-performing posts to compare."
# 5. Ask Mave for theme analysis + content generation
analysis = requests.post(f"{MV_BASE}/mave/chat", headers=MV_H, json={
"message": f"""Analyze these LinkedIn Company Page posts. I'm giving you the top 10 performers and bottom 5.
TOP PERFORMERS:
{top_block}
LOW PERFORMERS:
{bottom_block}
Produce:
1. THEME ANALYSIS: What do the top posts have in common? (format, length, hooks, tone, topics)
2. ANTI-PATTERNS: What do underperformers share?
3. CONTENT FORMULA: A repeatable structure based on the winners
4. NEW CONCEPTS: Generate 10 new LinkedIn post concepts that follow the winning patterns. For each: topic, hook (first line), format (text-only/carousel/video/poll), and why it should work based on the data.
5. POSTING CADENCE: Recommended frequency and timing based on content type distribution."""
}).json()
print("=" * 60)
print("LINKEDIN CONTENT STRATEGY — DATA-DRIVEN")
print("=" * 60)
print(f"Analyzed {len(enriched)} posts | Top engagement: {top[0]['engagement_score']}")
print(f"Avg engagement (top 10): {sum(p['engagement_score'] for p in top) / len(top):.0f}")
print(f"Avg engagement (all): {sum(p['engagement_score'] for p in enriched) / len(enriched):.0f}")
print()
print(analysis.get("content", "")[:2000])
