Competitor tracking

# competitor_tracking.py
# Snapshots one competitor on four platforms. Store the result and diff runs.
# Run with: INSIGHTSOCIAL_API_KEY=isk_live_... python competitor_tracking.py
import os
from concurrent.futures import ThreadPoolExecutor
from datetime import date

import requests

KEY = os.environ["INSIGHTSOCIAL_API_KEY"]
BASE = "https://api.insightsocial.app/v1"
COMPETITOR = "duolingo"  # the same handle on all four platforms here

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

PAIRS = {
    "tiktok":    ("tiktok/profile",    "tiktok/profile/videos",   {}),
    "instagram": ("instagram/profile", "instagram/profile/posts", {}),
    "youtube":   ("youtube/channel",   "youtube/channel/videos",  {"includeExtras": "true"}),
    "twitter":   ("twitter/profile",   "twitter/user/tweets",     {}),
}

def own_posts(items):
    # Leave out pinned posts (old, out of date order) and reposts of other accounts.
    posts = [i["post"] for i in items]
    return [p for p in posts
            if not p["flags"]["pinned"] and not (p["ext"] or {}).get("retweeted_post")]

def engagement_rate(posts, followers):
    """Average (likes + comments) / followers over the posts that show both counts."""
    if not followers:
        return None
    rates = [(p["engagement"]["likes"] + p["engagement"]["comments"]) / followers
             for p in posts
             if p["engagement"]["likes"] is not None and p["engagement"]["comments"] is not None]
    return round(sum(rates) / len(rates), 6) if rates else None

# Step 1: eight calls in parallel (the key allows 10 in flight).
with ThreadPoolExecutor(max_workers=8) as pool:
    futures = {}
    for platform, (profile_path, posts_path, extra) in PAIRS.items():
        futures[(platform, "profile")] = pool.submit(get, profile_path, handle=COMPETITOR)
        futures[(platform, "posts")] = pool.submit(get, posts_path, handle=COMPETITOR, **extra)
    results = {k: f.result() for k, f in futures.items()}

# Step 2: build today's snapshot.
snapshot = {"date": date.today().isoformat(), "handle": COMPETITOR, "platforms": []}
spent = 0
for platform in PAIRS:
    profile = results[(platform, "profile")]
    posts = results[(platform, "posts")]
    spent += profile.get("credits_used", 0) + posts.get("credits_used", 0)
    if not profile["success"]:
        print(f"{platform}: {profile['error']['type']}")
        continue
    author = profile["data"]["author"]
    items = own_posts(posts["data"]["items"]) if posts["success"] else []
    latest = items[0] if items else None
    snapshot["platforms"].append({
        "platform": platform,
        "followers": author["followers"],
        "followers_approximate": author["followers_approximate"],
        "engagement_rate": engagement_rate(items, author["followers"]),
        "posts_fetched": len(items),
        "latest_post_id": latest["id"] if latest else None,
        "latest_post_at": latest["published_at"] if latest else None,
        "latest_post": ((latest["content"]["text"] if latest else None) or "")[:80],
    })

for row in snapshot["platforms"]:
    print(row)
print(f"credits used this run: {spent}")

# Step 3: persist `snapshot` keyed by (handle, date), then compare with the last one:
#   follower_delta  = today.followers - previous.followers
#   new_posts       = posts published after previous.latest_post_at
#   engagement_jump = today.engagement_rate / trailing average
# Alert when a delta crosses your threshold.