Creator engagement scoring

# creator_scoring.py
# Run with: INSIGHTSOCIAL_API_KEY=isk_live_... python creator_scoring.py
import os
from concurrent.futures import ThreadPoolExecutor
from statistics import median

import requests

KEY = os.environ["INSIGHTSOCIAL_API_KEY"]
BASE = "https://api.insightsocial.app/v1"

# Real creators rarely share one handle, so map them per platform.
CREATOR = {"tiktok": "mrbeast", "instagram": "mrbeast", "youtube": "MrBeast"}
PATHS = {
    "tiktok":    ("tiktok/profile",    "tiktok/profile/videos",   {}),
    "instagram": ("instagram/profile", "instagram/profile/posts", {}),
    "youtube":   ("youtube/channel",   "youtube/channel/videos",  {"includeExtras": "true"}),
}

def get(path, **params):
    res = requests.get(f"{BASE}/{path}", params=params,
                       headers={"x-api-key": KEY}, timeout=120)
    return res.json()

def score(platform):
    profile_path, posts_path, extra = PATHS[platform]
    handle = CREATOR[platform]
    profile = get(profile_path, handle=handle)
    posts = get(posts_path, handle=handle, **extra)
    row = {"platform": platform, "followers": None, "engagement_rate": None, "posts_scored": 0,
           "credits": profile.get("credits_used", 0) + posts.get("credits_used", 0), "why": None}
    if not profile["success"] or not posts["success"]:
        row["why"] = (profile.get("error") or posts.get("error"))["type"]
        return row
    followers = profile["data"]["author"]["followers"]
    row["followers"] = followers
    if not followers:
        row["why"] = "no follower count"
        return row
    rates = []
    for item in posts["data"]["items"]:
        post = item["post"]
        likes, comments = post["engagement"]["likes"], post["engagement"]["comments"]
        # Skip pinned posts (old, out of order) and posts with a hidden count.
        if post["flags"]["pinned"] or likes is None or comments is None:
            continue
        rates.append((likes + comments) / followers)
    row["posts_scored"] = len(rates)
    if rates:
        row["engagement_rate"] = round(median(rates), 4)
    else:
        row["why"] = "no post with both counts"
    return row

with ThreadPoolExecutor(max_workers=3) as pool:
    rows = list(pool.map(score, CREATOR))

# Rank the comparable rows; keep the others visible but unranked.
ranked = sorted((r for r in rows if r["engagement_rate"] is not None),
                key=lambda r: r["engagement_rate"], reverse=True)
for r in ranked:
    print(f'{r["platform"]:<10} {r["followers"]:>12,}  {r["engagement_rate"]:.4f}  ({r["posts_scored"]} posts)')
for r in rows:
    if r["engagement_rate"] is None:
        print(f'{r["platform"]:<10} cannot compare: {r["why"]}')
print("credits used:", sum(r["credits"] for r in rows))