import os import re import streamlit as st import pandas as pd import plotly.express as px from googleapiclient.discovery import build from transformers import pipeline from typing import Union # Set environment variables for cache directories # This helps manage where Streamlit and Hugging Face cache files are stored, # especially useful in environments like Hugging Face Spaces. os.environ['STREAMLIT_SERVER_BROWSER_CACHE_PATH'] = '/tmp/.streamlit_browser_cache' os.environ['STREAMLIT_SERVER_LIB_CACHE_PATH'] = '/tmp/.streamlit_lib_cache' os.environ['STREAMLIT_SERVER_STATIC_CACHE_PATH'] = '/tmp/.streamlit_static_cache' os.environ['MPLCONFIGDIR'] = '/tmp/.config/matplotlib' # IMPORTANT: Changing this cache directory name (e.g., to _v5, _v6) # will force Hugging Face to re-download all models, effectively clearing # any potentially corrupted cached models. # Incrementing to _v6 to ensure a completely fresh start for the cache. os.environ['HUGGINGFACE_HUB_CACHE'] = '/tmp/huggingface_models_cache_v6' # Suppress TensorFlow OneDNN optimization warnings if TensorFlow is installed os.environ['TF_ENABLE_ONEDNN_OPTS'] = '0' # Configure Streamlit page settings st.set_page_config( layout="wide", page_title="YouTube Comment Sentiment Analyzer", page_icon="https://upload.wikimedia.org/wikipedia/commons/4/47/Google_pixel_logo.svg" ) # --- Debugger visualization (for environment variables) --- st.sidebar.subheader("Environment Variables (for debugging):") st.sidebar.write(f"HF_HOME: {os.environ.get('HF_HOME')}") st.sidebar.write(f"TRANSFORMERS_CACHE: {os.environ.get('TRANSFORMERS_CACHE')}") st.sidebar.write(f"HF_HUB_CACHE: {os.environ.get('HUGGINGFACE_HUB_CACHE')}") st.sidebar.write(f"YOUTUBE_API_KEY present: {bool(os.environ.get('YOUTUBE_API_KEY'))}") # --- End debugger --- # Regex to extract YouTube video ID from various URL formats YOUTUBE_URL_REGEX = re.compile( r"(?:youtube\.com\/(?:[^\/]+\/.+\/|(?:v|e(?:mbed)?)\/|.*[?&]v=)|youtu\.be\/)([^\"&?\/\s]{11})" ) def get_video_id(youtube_url: str) -> Union[str, None]: """ Extracts the YouTube video ID from a given URL. Args: youtube_url (str): The URL of the YouTube video. Returns: Union[str, None]: The 11-character video ID if found, otherwise None. """ match = YOUTUBE_URL_REGEX.search(youtube_url) return match.group(1) if match else None def fetch_comments(video_id: str, api_key: str, max_comments: int = 500) -> list[str]: """ Fetches comments for a given YouTube video ID using the YouTube Data API. Args: video_id (str): The ID of the YouTube video. api_key (str): Your Google/YouTube Data API key. max_comments (int): The maximum number of comments to fetch. Returns: list[str]: A list of fetched comment texts. """ try: # Initialize YouTube API service client youtube = build("youtube", "v3", developerKey=api_key) comments = [] next_page_token = None fetched_count = 0 # Loop to fetch comments page by page until max_comments is reached or no more pages while fetched_count < max_comments: request = youtube.commentThreads().list( part="snippet", videoId=video_id, maxResults=100, # Max results per page allowed by API pageToken=next_page_token, textFormat="plainText" ) response = request.execute() # Extract comment text from the response for item in response.get("items", []): comment_text = item["snippet"]["topLevelComment"]["snippet"]["textDisplay"] comments.append(comment_text) fetched_count += 1 if fetched_count >= max_comments: break # Get token for the next page of comments next_page_token = response.get("nextPageToken") if not next_page_token: break # No more pages return comments except Exception as e: # Log the error for debugging in the console/logs print(f"Error fetching comments: {e}") # Display user-friendly error messages in Streamlit st.error(f"An error occurred while fetching comments: {e}") if "disabled comments" in str(e).lower(): st.info("Comments might be disabled for this video, or it's a Live Chat replay.") elif "quota" in str(e).lower(): st.info("You might have hit your YouTube Data API daily quota.") return [] # Sentiment analysis model name. This is the standard DistilBERT SST-2 finetuned model. # If you intended to use the 'tiny' version, change this to "sshleifer/tiny-distilbert-base-uncased-finetuned-sst-2-english" SENTIMENT_MODEL_NAME = "distilbert-base-uncased-finetuned-sst-2-english" # Using the explicitly set environment for my cache directory. # In case of errors, fallback to a default /tmp path. HF_CACHE_DIR = os.environ.get('HUGGINGFACE_HUB_CACHE', '/tmp/huggingface_models_cache') @st.cache_resource def load_sentiment_model(): """ Loads the sentiment analysis model using the Hugging Face transformers pipeline. This function is cached by Streamlit to avoid re-loading the model on every rerun. """ try: # Initialize the sentiment analysis pipeline sentiment_pipeline = pipeline( "sentiment-analysis", model=SENTIMENT_MODEL_NAME, tokenizer=SENTIMENT_MODEL_NAME, top_k=None, # Get scores for all labels (POSITIVE, NEGATIVE) # The HUGGINGFACE_HUB_CACHE environment variable is automatically respected by pipeline # Removed force_download and local_files_only to simplify and rely on cache path change. ) return sentiment_pipeline except Exception as e: # Provide detailed error messages to the user st.error(f"Error loading sentiment model: {e}") st.info(f"The app attempted to load model: '{SENTIMENT_MODEL_NAME}'.") st.info(f"Please ensure the model '{SENTIMENT_MODEL_NAME}' is available and compatible.") st.info(f"The model was attempted to be cached to: `{HF_CACHE_DIR}`.") st.info("If the issue persists, consider these steps:") st.info("1. **Check your internet connection** to ensure a stable download.") st.info(f"2. **Force a fresh download** by changing the `HUGGINGFACE_HUB_CACHE` env var in your Space settings or directly in the `streamlit_app.py` file (e.g., to `/tmp/huggingface_models_cache_v7`).") st.info("3. **Verify model availability** on Hugging Face Hub: `https://huggingface.co/{SENTIMENT_MODEL_NAME}`.") st.stop() # Stop the Streamlit app execution gracefully if the model can't load def analyze_sentiments(comments: list[str], sentiment_pipeline) -> pd.DataFrame: """ Analyzes the sentiment of a list of comments using the provided sentiment pipeline. Args: comments (list[str]): A list of comment texts. sentiment_pipeline: The loaded Hugging Face sentiment analysis pipeline. Returns: pd.DataFrame: A DataFrame containing comments, their predicted sentiment, and score. """ if not comments: return pd.DataFrame() comment_data = [] for comment_text in comments: # Skip empty or whitespace-only comments if not comment_text.strip(): continue sentiment_results = sentiment_pipeline(comment_text) # Extract the main sentiment (POSITIVE/NEGATIVE) and its score if sentiment_results and sentiment_results[0]: # The model returns a list of dictionaries, usually sorted by score. # We take the first one as the most confident prediction. main_sentiment = sentiment_results[0][0]['label'] score = sentiment_results[0][0]['score'] comment_data.append({ "Comment": comment_text, "Sentiment": main_sentiment, "Score": round(score, 4) }) return pd.DataFrame(comment_data) def plot_sentiment_distribution(sentiment_df: pd.DataFrame): """ Plots the distribution of sentiments using a bar chart. Args: sentiment_df (pd.DataFrame): DataFrame containing sentiment analysis results. """ if sentiment_df.empty: st.warning("No sentiment data to plot distribution.") return # Calculate counts for each sentiment label sentiment_counts = sentiment_df['Sentiment'].value_counts().reset_index() sentiment_counts.columns = ['Sentiment', 'Count'] # Define colors for the plot (Distilbert-base-uncased-finetuned-sst-2-english is binary) color_map = {'POSITIVE': 'green', 'NEGATIVE': 'red'} fig = px.bar( sentiment_counts, x='Sentiment', y='Count', title='Distribution of Comment Sentiments', color='Sentiment', color_discrete_map=color_map, text='Count' # Display count value on top of bars ) # Update layout for better readability fig.update_traces(texttemplate='%{text}', textposition='outside') fig.update_layout(xaxis_title="Sentiment", yaxis_title="Number of Comments") st.plotly_chart(fig, use_container_width=True) def display_sentiment_scores(sentiment_df: pd.DataFrame): """ Displays individual comment sentiments in a Streamlit DataFrame. Args: sentiment_df (pd.DataFrame): DataFrame containing sentiment analysis results. """ if sentiment_df.empty: st.info("No comments available to display individual sentiment scores.") return st.subheader("Individual Comment Sentiments:") st.dataframe(sentiment_df, use_container_width=True) # --- Main Application Logic --- YOUTUBE_API_KEY = os.environ.get("YOUTUBE_API_KEY") st.title("🎥 (Phoenix) YouTube Comment Sentiment Checker") st.markdown("Have you been attacked on YouTube? Report this by entering the YouTube video URL to fetch its comments and analyze their sentiment (Positive/Negative).") # Check if API key is set if not YOUTUBE_API_KEY: st.warning("🚨 **YouTube API Key not found!** Please add `YOUTUBE_API_KEY` to your Hugging Face Space secrets or environment variables.") st.info("Instructions on how to add secrets to Hugging Face Spaces can be found in their documentation.") st.stop() # Stop execution if API key is missing # Load the sentiment model with a spinner with st.spinner("🚀 Loading sentiment model... This might take a moment if it's the first run (downloading model)."): sentiment_classifier = load_sentiment_model() # User input for YouTube URL youtube_url = st.text_input( "🔗 Enter YouTube Video URL:", "https://www.youtube.com/watch?v=dQw4w9WgXcQ", # Example URL (Rick Astley - Never Gonna Give You Up) help="e.g., https://www.youtube.com/watch?v=dQw4w9WgXcQ" ) # Slider for maximum comments to fetch max_comments = st.slider( "📊 Max Comments to Fetch (approx.):", min_value=50, max_value=2000, value=500, step=50, help="Fetching more comments takes longer and uses more API quota." ) # Button to trigger analysis if st.button("✨ Analyze Comments", type="primary"): if not youtube_url: st.error("❗ Please enter a YouTube video URL to analyze.") else: video_id = get_video_id(youtube_url) if not video_id: st.error("❌ Invalid YouTube URL. Please make sure it's a valid video URL (e.g., starts with youtube.com/watch?v= or youtu.be/).") else: st.info(f"Fetching comments for video ID: `{video_id}`") # Fetch comments with a spinner with st.spinner(f"📥 Fetching up to {max_comments} comments from YouTube..."): comments = fetch_comments(video_id, YOUTUBE_API_KEY, max_comments) if not comments: st.warning("⚠️ No comments found for this video, or an error occurred while fetching. Ensure comments are enabled and public.") else: st.success(f"✅ Successfully fetched {len(comments)} comments.") # Analyze sentiments with a spinner with st.spinner("🧠 Analyzing sentiment of comments..."): sentiment_df = analyze_sentiments(comments, sentiment_classifier) if sentiment_df.empty: st.info("No comments were successfully analyzed for sentiment.") else: # Display results st.subheader("📈 Sentiment Analysis Results:") # Display summary statistics total_analyzed = len(sentiment_df) positive_count = sentiment_df[sentiment_df['Sentiment'] == 'POSITIVE'].shape[0] negative_count = sentiment_df[sentiment_df['Sentiment'] == 'NEGATIVE'].shape[0] st.write(f"**Total comments analyzed:** `{total_analyzed}`") st.write(f"**Positive sentiments:** `{positive_count}` ({(positive_count/total_analyzed*100):.1f}%)") st.write(f"**Negative sentiments:** `{negative_count}` ({(negative_count/total_analyzed*100):.1f}%)") # Plot sentiment distribution plot_sentiment_distribution(sentiment_df) # Display individual sentiment scores display_sentiment_scores(sentiment_df) st.markdown("---") st.markdown("This is a Build Up project developed last night with ❤️ as an example of a modular addition to support the reporting of TFGBV platforms in Kenya .")