Blame for src/streamlit_app.py

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 .")