Add main.py, yt_api.py, requirements.txt

main.py

@@ -0,0 +1,25 @@
+from fastapi import FastAPI
+from yt_api import get_comments
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    YT_API_KEY: str
+    model_config = SettingsConfigDict(env_file='.env')
+
+
+settings = Settings()
+app = FastAPI(title='social-stat')
+
+
+YT_API_KEY = settings.YT_API_KEY
+
+
+@app.get('/')
+def home():
+    return 'social-stat'
+
+
+@app.post('/predict')
+def predict(video_id):
+    return get_comments(video_id, YT_API_KEY)

requirements.txt

@@ -0,0 +1,4 @@
+requests
+fastapi
+uvicorn
+pydantic_settings

yt_api.py

@@ -0,0 +1,62 @@
+import requests
+# from pprint import pprint
+
+
+def get_comments(video_id, api_key):
+    """Yields all `commentThreads` from a YouTube video in batches."""
+
+    # Get comments from the first page.
+    response = get_response(video_id, api_key, max_results=100)
+    comment_list = response_to_comments(response)
+
+    # Get comments from the other pages.
+    while 'nextPageToken' in response.keys():
+        response = get_response(
+            video_id, api_key, page_token=response['nextPageToken'])
+        comment_list.update(response_to_comments(response))
+
+    return comment_list
+
+
+def get_response(video_id, api_key, page_token=None, max_results=100):
+    """Gets the response from YouTube API and converts it to JSON."""
+    url = 'https://youtube.googleapis.com/youtube/v3/commentThreads'
+    payload = {
+        'videoId': video_id,
+        'key': api_key,
+        'maxResults': max_results,
+        'part': 'snippet',
+        'pageToken': page_token,
+    }
+    response = requests.get(url, params=payload)
+    return response.json()
+
+
+def response_to_comments(response):
+    """Converts JSON response to `comment_list` dict."""
+    comment_list = {}
+    for comment in response['items']:
+        comment = comment['snippet']
+        can_reply = comment['canReply']
+        total_reply_count = comment['totalReplyCount']
+        comment = comment['topLevelComment']
+        comment_id = comment['id']
+        comment = comment['snippet']
+        try:
+            comment_list[comment_id] = {
+                'video_id': comment['videoId'],
+                'channel_id': comment['authorChannelId']['value'],
+                'author_display_name': comment['authorDisplayName'],
+                'text_original': comment['textOriginal'],
+                'text_display': comment['textDisplay'],
+                'published_at': comment['publishedAt'].replace('T', ' ')[:-1],
+                'updated_at': comment['updatedAt'].replace('T', ' ')[:-1],
+                'like_count': comment['likeCount'],
+                'can_reply': can_reply,
+                'total_reply_count': total_reply_count,
+            }
+        except Exception as e:
+            print(f"Error: {e}\nComment: {comment}")
+            continue
+
+    return comment_list