Sentiment analysis

# sentiment.py
# Collects comments about a topic and counts the sentiment split.
# Run with: INSIGHTSOCIAL_API_KEY=isk_live_... python sentiment.py
import os
from collections import Counter
from concurrent.futures import ThreadPoolExecutor

import requests
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer

KEY = os.environ["INSIGHTSOCIAL_API_KEY"]
BASE = "https://api.insightsocial.app/v1"
TOPIC = "github copilot pricing"
KEYWORDS = ["copilot"]  # a result must mention one of these to count as on-topic
THREADS_PER_PLATFORM = 3

analyzer = SentimentIntensityAnalyzer()

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

def on_topic(post):
    text = (post["content"]["text"] or "").lower()
    return any(k in text for k in KEYWORDS)

def band(text):
    # VADER's usual cut-offs on the compound score, from -1 to 1.
    score = analyzer.polarity_scores(text)["compound"]
    return "negative" if score <= -0.05 else "positive" if score >= 0.05 else "neutral"

def flatten_thread(comments):
    # Reddit nests replies inside each comment; other platforms leave `replies` null.
    for c in comments:
        yield c
        yield from flatten_thread(c["replies"] or [])

# Step 1: find the conversation on both platforms, keeping on-topic results.
yt = get("youtube/search", query=TOPIC)
rd = get("reddit/search", query=TOPIC, timeframe="month")

def top_urls(body, keep=lambda url: True):
    if not body["success"]:
        return []
    posts = [i["post"] for i in body["data"]["items"]]
    return [p["url"] for p in posts if p["url"] and keep(p["url"]) and on_topic(p)][:THREADS_PER_PLATFORM]

videos = top_urls(yt, keep=lambda u: "watch?v=" in u)
threads = top_urls(rd)

# Step 2: pull the comments under the top results.
jobs = [("youtube/video/comments", u) for u in videos] + \
       [("reddit/post/comments", u) for u in threads]

with ThreadPoolExecutor(max_workers=6) as pool:
    pages = list(pool.map(lambda j: (j, get(j[0], url=j[1])), jobs))

# Step 3: score every comment and count the split per platform.
split = {"youtube": Counter(), "reddit": Counter()}
spent = yt.get("credits_used", 0) + rd.get("credits_used", 0)
for (path, url), body in pages:
    spent += body.get("credits_used", 0)
    if not body["success"]:
        continue
    platform = path.split("/")[0]
    for c in flatten_thread(item["comment"] for item in body["data"]["items"]):
        # The scorer is English-only. `language` is often null; keep those.
        if not c["text"] or c["language"] not in (None, "en"):
            split[platform]["skipped"] += 1
            continue
        split[platform][band(c["text"])] += 1

for platform, counts in split.items():
    print(platform, dict(counts))
print(f"credits used: {spent}")