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"),
    "twitter":   ("twitter/profile",   "twitter/user/tweets"),
}

# Step 1: eight calls in parallel (the key allows 10 in flight).
with ThreadPoolExecutor(max_workers=8) as pool:
    futures = {
        (platform, kind): pool.submit(get, path, handle=COMPETITOR)
        for platform, paths in PAIRS.items()
        for kind, path in zip(("profile", "posts"), paths)
    }
    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
    items = posts.get("data", {}).get("items", []) if posts["success"] else []
    latest = items[0]["post"] if items else {}
    snapshot["platforms"].append({
        "platform": platform,
        "followers": profile["data"]["author"].get("followers"),
        "engagement_rate": (profile["data"].get("computed") or {}).get("engagement_rate"),
        "posts_fetched": len(items),
        "latest_post": (latest.get("text") 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 the previous run
#   engagement_jump = today.engagement_rate / trailing average
# Alert when a delta crosses your threshold.