| | import os
|
| | import googleapiclient.discovery
|
| | import googleapiclient.errors
|
| | from dotenv import load_dotenv
|
| |
|
| |
|
| | import streamlit as st
|
| |
|
| | load_dotenv()
|
| | api_key = os.getenv("API_KEY")
|
| |
|
| |
|
| |
|
| | def get_comments(youtube, **kwargs):
|
| | comments = []
|
| | results = youtube.commentThreads().list(**kwargs).execute()
|
| |
|
| | while results:
|
| | for item in results["items"]:
|
| | comment = item["snippet"]["topLevelComment"]["snippet"]["textDisplay"]
|
| | comments.append(comment)
|
| |
|
| |
|
| | if "nextPageToken" in results:
|
| | kwargs["pageToken"] = results["nextPageToken"]
|
| | results = youtube.commentThreads().list(**kwargs).execute()
|
| | else:
|
| | break
|
| |
|
| | return comments
|
| |
|
| |
|
| | def main(video_id, api_key):
|
| |
|
| | os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"
|
| |
|
| | youtube = googleapiclient.discovery.build("youtube", "v3", developerKey=api_key)
|
| |
|
| | video_title = "N/A"
|
| |
|
| | try:
|
| |
|
| | print(f"DEBUG (youtube.py): Fetching video details for ID: {video_id}")
|
| | video_response = (
|
| | youtube.videos()
|
| | .list(
|
| | part="snippet",
|
| | id=video_id,
|
| | )
|
| | .execute()
|
| | )
|
| |
|
| |
|
| |
|
| | if video_response.get("items"):
|
| | video_title = video_response["items"][0]["snippet"]["title"]
|
| | print(f"DEBUG (youtube.py): Found title: '{video_title}'")
|
| | else:
|
| | print(f"WARN (youtube.py): No video items found for ID: {video_id}")
|
| | video_title = "Video Not Found or Private"
|
| |
|
| | except Exception as e:
|
| | print(
|
| | f"ERROR (youtube.py): Failed to fetch video title for ID {video_id}. Error: {e}"
|
| | )
|
| | video_title = "Error Fetching Title"
|
| |
|
| | comments = get_comments(
|
| | youtube, part="snippet", videoId=video_id, textFormat="plainText"
|
| | )
|
| |
|
| |
|
| | return {"title": video_title, "comments": comments}
|
| |
|
| |
|
| | def get_video_comments(video_id):
|
| | return main(video_id, api_key)
|
| |
|