app.py

import gradio as gr
from transformers import pipeline
from wordcloud import WordCloud, STOPWORDS
from youtubesearchpython import *
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from PIL import Image
import re
import io
from io import BytesIO
import time


sentiment_task = pipeline("sentiment-analysis", model="cardiffnlp/twitter-roberta-base-sentiment-latest", tokenizer="cardiffnlp/twitter-roberta-base-sentiment-latest")

def extract_youtube_video_id(url_or_id):
    """
    Extracts the YouTube video ID from a given URL or returns the ID if a direct ID is provided.
    Args:
    url_or_id (str): A YouTube URL or a video ID.
    Returns:
    str: The extracted YouTube video ID.
    """
    # Check if it's already a valid YouTube ID (typically 11 characters)
    if len(url_or_id) == 11 and not re.search(r'[^0-9A-Za-z_-]', url_or_id):
        return url_or_id

    # Regular expressions for various YouTube URL formats
    regex_patterns = [
        r'(?:https?://)?www\.youtube\.com/watch\?v=([0-9A-Za-z_-]{11})',
        r'(?:https?://)?youtu\.be/([0-9A-Za-z_-]{11})',
        r'(?:https?://)?www\.youtube\.com/embed/([0-9A-Za-z_-]{11})',
        r'(?:https?://)?www\.youtube\.com/v/([0-9A-Za-z_-]{11})',
        r'(?:https?://)?www\.youtube\.com/shorts/([0-9A-Za-z_-]{11})'
    ]

    # Try each regex pattern to find a match
    for pattern in regex_patterns:
        match = re.search(pattern, url_or_id)
        if match:
            return match.group(1)

    # If no pattern matches, return an error or a specific message
    return "Invalid YouTube URL or ID"

def comments_collector(video_link, max_comments = 100):
  # This function collects comments from a given YouTube video link.
  # It uses the youtubesearchpython library to extract comments and pandas for data manipulation.
  # Args:
  #   video_link (str): The YouTube video link from which to collect comments.
  #   max_comments (int, optional): The maximum number of comments to retrieve. Defaults to 100.
  # Returns:
  #   pandas.DataFrame: A DataFrame containing the comments, or None in case of an exception.
  video_id = extract_youtube_video_id(video_link)
  max_comments -= 1

  try:
    #load the first 20 comments
    comments = Comments(video_id)
    print(f'Comments Retrieved and Loading...')

    #load more comments, 20 at a time, until the limit is reached
    while comments.hasMoreComments and (len(comments.comments["result"]) <= max_comments):
      comments.getNextComments()
    print(f'Found all the {len(comments.comments["result"])} comments.')

    #load all the comments into "comments" variable
    comments = comments.comments

    #define data list for collecting comments for a particular video
    data = []

    #loop through all the comments
    for i in range(len(comments['result'])):
      #############################################################################
      is_author = comments['result'][i]['authorIsChannelOwner']

      #check if the comment is from the video author or not -> neglect if so.
      if is_author:
        pass
      #############################################################################
      #comment comes from others, we will save this comment.
      else:
        comment_dict = {}
        comment_id = comments['result'][i]['id']
        author = comments['result'][i]['author']['name']
        content = comments['result'][i]['content']

        #############################################################################
        #cleaning comments likes (e.g., convert 10K likes to 10000, convert None like to 0)
        if comments['result'][i]['votes']['label'] is None:
          likes = 0
        else:
          likes = comments['result'][i]['votes']['label'].split(' ')[0]
          if 'K' in likes:
            likes = int(float(likes.replace('K', '')) * 1000)

        #############################################################################
        #cleaning comments reply count
        replyCount = comments['result'][i]['replyCount']
        #if there is no reply, we will log it as 0
        if replyCount is None:
          comment_dict['replyCount'] = 0
        #otherwise, we will log as integer
        else:
          comment_dict['replyCount'] = int(replyCount)

        #############################################################################
        comment_dict['comment_id'] = comment_id
        comment_dict['author'] = author
        comment_dict['content'] = content
        comment_dict['likes'] = likes

        data.append(comment_dict)
        #############################################################################
    print(f'Excluding author comments, we ended up with {len(data)} comments')
    return pd.DataFrame(data)
  except Exception as e:
    print(e)
    return None

def comments_analyzer(comments_df):
  # This function analyzes the sentiment of comments in a given DataFrame.
  # It requires a DataFrame of comments, typically generated by the comments_collector function.
  # Args:
  #   comments_df (pandas.DataFrame): A DataFrame containing YouTube comments.
  # Returns:
  #   dict: A dictionary with analysis results, including sentiment counts and percentages, or None if input is None.
  # The function applies a sentiment analysis task on each comment and categorizes them as positive, neutral, or negative.
  # It also calculates the percentage of positive comments and blends all comments into a single string.
  if comments_df is None:
    return None
  else:

    start_time = time.time()
    # Example of batch processing with sentiment and confidence
    batch_size = 20  # Adjust the size based on your system's capabilities
    sentiments = []
    scores = []

    for i in range(0, len(comments_df), batch_size):
        batch = comments_df['content'][i:i+batch_size].tolist()
        batch_results = sentiment_task(batch)
        
        # Extracting both sentiment labels and scores
        batch_sentiments = [item['label'] for item in batch_results]
        batch_scores = [item['score'] for item in batch_results]

        sentiments.extend(batch_sentiments)
        scores.extend(batch_scores)

    comments_df['sentiment'] = sentiments
    comments_df['score'] = scores

    end_time = time.time()
    print(f"Time taken for batch sentiment analysis: {end_time - start_time} seconds")

    def get_top_comments(comments, sentiment_type, top_n=3):
        filtered_comments = comments[comments['sentiment'] == sentiment_type]
        top_comments = filtered_comments.nlargest(top_n, 'score')

        if not top_comments.empty:
            return '\n\n'.join(f"{row['content']} - {row['author']}" for _, row in top_comments.iterrows())
        else:
            return f"No {sentiment_type} comments available."

    start_time = time.time()
    # Get top positive comments
    top_positive_comments = get_top_comments(comments_df, 'positive')

    # Get top negative comments
    top_negative_comments = get_top_comments(comments_df, 'negative')
    end_time = time.time()
    print(f"Time taken for finding top n positive/negative comments: {end_time - start_time} seconds")

    data = {}
    #Categorize the comments by sentiment and count them
    data['total_comments'] = len(comments_df)
    data['num_positive'] = comments_df['sentiment'].value_counts().get('positive', 0)
    data['num_neutral'] = comments_df['sentiment'].value_counts().get('neutral', 0)
    data['num_negative'] = comments_df['sentiment'].value_counts().get('negative', 0)

    #blend all the comments
    data['blended_comments'] = comments_df['content'].str.cat(sep=' ')
    data['pct_positive'] = 100 * round(data['num_positive']/data['total_comments'], 2)

  return data, top_positive_comments, top_negative_comments

def generate_wordcloud(long_text, additional_stopwords=['Timestamps', 'timestamps']):
  # This function generates a word cloud image from a given text and returns it as a PIL image object.
  # Args:
  #   long_text (str): The text from which to generate the word cloud.
  #   additional_stopwords (list, optional): A list of words to be excluded from the word cloud.
  # The function creates a word cloud with specified font size, word limit, and background color.
  # It then converts the matplotlib plot to a PIL Image object for further use or saving.
  # Returns:
  #   PIL.Image: The generated word cloud as a PIL image object.

  #Call the default STOPWORDS from wordcloud library
  stopwords = set(STOPWORDS)

  #Combine the default STOPWORDS with the manually specified STOPWORDS to exclude them from the wordcloud.
  all_stopwords = stopwords.union(additional_stopwords)

  # Create a Word Cloud
  wordcloud = WordCloud(max_font_size=50, max_words=20, background_color="black", stopwords=all_stopwords, colormap='plasma').generate(long_text)

  # Create a figure
  plt.figure(figsize=(10,10), facecolor=None)
  plt.imshow(wordcloud, interpolation="bilinear")
  plt.axis("off")
  plt.tight_layout(pad=0)

  # Save to a BytesIO object
  img_buf = io.BytesIO()
  plt.savefig(img_buf, format='png', bbox_inches='tight', pad_inches=0)
  img_buf.seek(0)

  # Close the plt figure to prevent display
  plt.close()

  # Use PIL to open the image from the BytesIO object
  image = Image.open(img_buf)

  return image

def create_sentiment_analysis_chart(data):
  # This function creates a bar chart for sentiment analysis results and returns it as a PIL image object.
  # Args:
  #   data (dict): A dictionary containing the count of positive, negative, and neutral comments.
  # The function first converts the input data into a pandas DataFrame.
  # It then creates a bar chart using matplotlib, setting specific colors for different sentiment types.
  # Titles, labels, and legends are added for clarity.
  # Finally, the plot is saved to a BytesIO object and converted to a PIL image.
  # Returns:
  #   PIL.Image: The sentiment analysis bar chart as a PIL image object.

  # Convert the data to a DataFrame
  df = {}
  df['num_positive'] = data['num_positive']
  df['num_negative'] = data['num_negative']
  df['num_neutral'] = data['num_neutral']
  df = pd.DataFrame(df, index=[0])

  # Plotting
  plt.figure(figsize=(8, 6))
  bar_colors = ['green', 'red', 'blue']  # Colors for positive, negative, neutral
  df.plot(kind='bar', color=bar_colors, legend=True)

  # Adding titles and labels
  plt.title('Sentiment Analysis Results')
  plt.xlabel('Sentiment Types')
  plt.ylabel('Number of Comments')
  plt.xticks(ticks=[0], labels=['Sentiments'], rotation=0)  # Adjust x-ticks
  plt.legend(['Positive', 'Negative', 'Neutral'])

  # Save the plot to a BytesIO object
  buf = BytesIO()
  plt.savefig(buf, format='png')
  buf.seek(0)

  # Close the plt figure to prevent display
  plt.close()

  # Use PIL to open the image from the BytesIO object
  image = Image.open(buf)

  return image


############################################################################################################################################
# The code for processing the YouTube link, generating the word cloud, summary, and sentiment analysis
# should be defined here (using your existing functions).

def process_youtube_comments(youtube_link, max_comments, stop_words):
    # Process the YouTube link and generate the word cloud, summary, and sentiment analysis

    start_time = time.time()

    # Pull comments from the YouTube Video
    comments_df = comments_collector(video_link=youtube_link, max_comments=max_comments)

    end_time = time.time()
    print(f"Time taken for loading comments: {end_time - start_time} seconds")
    
    # Analyze
    analysis_dict, top_positive_comments, top_negative_comments = comments_analyzer(comments_df)

    long_text = analysis_dict['blended_comments']

    start_time = time.time()
    
    # Generate word cloud
    word_cloud_img = generate_wordcloud(long_text, additional_stopwords=['Timestamps', 'timestamps'])

    end_time = time.time()
    print(f"Time taken for generating word clouds: {end_time - start_time} seconds")

    start_time = time.time()

    # Create Sentiment Chart
    sentiment_chart = create_sentiment_analysis_chart(analysis_dict)

    end_time = time.time()
    print(f"Time taken for creating sentiment chart: {end_time - start_time} seconds")

    # Return the generated word cloud image, summary text, and sentiment analysis chart
    return word_cloud_img, top_positive_comments, top_negative_comments, sentiment_chart

############################################################################################################################################
# Gradio interface
interface = gr.Interface(
    fn=process_youtube_comments,
    inputs=[
        gr.Textbox(label="YouTube Video Link"),
        gr.Number(label="Maximum Comments", value=100),
        gr.Textbox(label="Words to exclude from cloud (comma-separated)")
    ],
    outputs=[
        gr.Image(label="Word Cloud ☁️"),
        gr.Textbox(label="Top 3 Positive Comments 👍🏻"),
        gr.Textbox(label="Top 3 Negative Comments 👎🏻"),
        gr.Image(label="Sentiment Analysis Chart 📊")
    ],
    title="YouTube Comments Analyzer 📈",
    description="Enter a YouTube link to generate a word cloud, top positive and negative comments, and sentiment analysis of the comments. \n\n Note: The app is both desktop 🖥️/mobile 📱 compatible. Depending on the amount of comments found, it can take up to 1 - 2 mins to process. Have fun 😀!"
)

# Run the interface
interface.launch()
############################################################################################################################################