YouTube Comment Intelligence with YouTube Data API v3, Streamlit, and Plotly

1from googleapiclient.discovery import build
2import time
3from typing import Generator
4
5class YouTubeCommentsExporter:
6    def __init__(self, api_key: str):
7        self.youtube = build("youtube", "v3", developerKey=api_key)
8        self.quota_used = 0
9
10    def export_all_comments(self, video_id: str) -> Generator[dict, None, None]:
11        """Export all comments with pagination. Yields comment dicts."""
12        next_page_token = None
13
14        while True:
15            # Each list call costs 1 quota unit
16            response = self.youtube.commentThreads().list(
17                part="snippet",
18                videoId=video_id,
19                maxResults=100,
20                pageToken=next_page_token,
21                textFormat="plainText",
22                order="relevance",
23            ).execute()
24            self.quota_used += 1
25
26            for item in response["items"]:
27                comment = item["snippet"]["topLevelComment"]["snippet"]
28                yield {
29                    "comment_id": item["id"],
30                    "author": comment["authorDisplayName"],
31                    "text": comment["textDisplay"],
32                    "likes": comment["likeCount"],
33                    "published_at": comment["publishedAt"],
34                    "reply_count": item["snippet"]["totalReplyCount"],
35                }
36
37            next_page_token = response.get("nextPageToken")
38            if not next_page_token:
39                break
40
41            # Rate limit: 100 requests/100 seconds
42            if self.quota_used % 90 == 0:
43                time.sleep(1)

1import re
2from sklearn.feature_extraction.text import TfidfVectorizer
3from sklearn.cluster import KMeans
4
5class CommentClassifier:
6    QUESTION_PATTERNS = [
7        r"\bwhy\b", r"\bhow\b", r"\bwhat\b", r"\bwhen\b",
8        r"\bwhere\b", r"\bcan you\b", r"\bdo you\b", r"\?",
9    ]
10
11    SPAM_PATTERNS = [
12        r"check out my channel", r"subscribe to me",
13        r"I make \$\d+", r"bit\.ly/", r"free.*giveaway",
14    ]
15
16    def classify(self, df) -> dict:
17        # Question detection — content ideas goldmine
18        question_pattern = "|".join(self.QUESTION_PATTERNS)
19        df["is_question"] = df["text"].str.lower().str.contains(
20            question_pattern, regex=True
21        )
22
23        # Spam filtering
24        spam_pattern = "|".join(self.SPAM_PATTERNS)
25        df["is_spam"] = df["text"].str.lower().str.contains(
26            spam_pattern, regex=True
27        )
28
29        # Sentiment scoring (simple lexicon-based, no API cost)
30        positive_words = {"great", "amazing", "love", "helpful", "awesome", "thank"}
31        negative_words = {"bad", "wrong", "hate", "useless", "boring", "worse"}
32
33        def sentiment_score(text):
34            words = set(text.lower().split())
35            pos = len(words & positive_words)
36            neg = len(words & negative_words)
37            if pos > neg: return "positive"
38            if neg > pos: return "negative"
39            return "neutral"
40
41        df["sentiment"] = df["text"].apply(sentiment_score)
42        return df
43
44    def cluster_topics(self, texts: list[str], n_clusters: int = 5) -> list[int]:
45        """Cluster comments by topic using TF-IDF + KMeans."""
46        vectorizer = TfidfVectorizer(
47            max_features=500,
48            stop_words="english",
49            ngram_range=(1, 2),
50        )
51        X = vectorizer.fit_transform(texts)
52        km = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
53        return km.fit_predict(X).tolist()

1import pandas as pd
2import plotly.graph_objects as go
3from datetime import datetime, timedelta
4
5def compute_view_velocity(video_stats: list[dict]) -> pd.DataFrame:
6    """
7    Computes view velocity: views acquired in the first N hours vs. later.
8    Uses video publish time + current view count (YouTube API doesn't give
9    hourly breakdown, so we use time-since-publish as a proxy).
10    """
11    now = datetime.now(tz=timezone.utc)
12    df = pd.DataFrame(video_stats)
13    df["published_at"] = pd.to_datetime(df["published_at"])
14    df["hours_since_publish"] = (now - df["published_at"]).dt.total_seconds() / 3600
15    df["views_per_hour"] = df["view_count"] / df["hours_since_publish"].clip(lower=1)
16
17    # Categorize: hot (< 48h), warm (2-7 days), cold (> 7 days)
18    df["velocity_phase"] = pd.cut(
19        df["hours_since_publish"],
20        bins=[0, 48, 168, float("inf")],
21        labels=["hot (< 48h)", "warm (2-7 days)", "cold (> 7 days)"],
22    )
23    return df
24
25def best_upload_day(df: pd.DataFrame) -> dict:
26    """Find best day to upload based on first-48h view velocity."""
27    df["upload_day"] = df["published_at"].dt.day_name()
28    hot_videos = df[df["velocity_phase"] == "hot (< 48h)"]
29    return (
30        hot_videos.groupby("upload_day")["views_per_hour"]
31        .mean()
32        .sort_values(ascending=False)
33        .to_dict()
34    )