I have limited resourced to do ML ion my MacBook. its not M1 or m2, therefore I can't use LLM to classify YouTube comments. so I want to try RAG and distilbert.
check the script:
import logging
import re
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import DistilBertForSequenceClassification, DistilBertTokenizer
import torch
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from config import YOUTUBE_API_KEY  # Importing the API key from config.py

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Define a FastAPI application
app = FastAPI(
    title="YouTube Video Emotion Analyzer",
    description="Analyze emotions in YouTube video comments using DistilBERT.",
    version="1.0.0"
)

# Define a Pydantic model for input data
class EmotionRequest(BaseModel):
    video_id: str  # Accepts the video ID from the request body

class EmotionAnalyzer:
    def __init__(self, model_name='distilbert-base-uncased-finetuned-sst-2-english', device=None):
        logger.info(f"Initializing EmotionAnalyzer with model: {model_name}")

        # Set device
        if device:
            self.device = device
        else:
            self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        logger.info(f"Using device: {self.device}")

        # Load the DistilBERT model and tokenizer
        try:
            logger.info("Loading DistilBERT model...")
            self.model = DistilBertForSequenceClassification.from_pretrained(model_name)
            self.model.to(self.device)
            logger.info("Model loaded successfully.")
        except Exception as e:
            logger.error(f"Error loading model: {e}")
            raise

        try:
            logger.info("Loading tokenizer...")
            self.tokenizer = DistilBertTokenizer.from_pretrained(model_name)
            logger.info("Tokenizer loaded successfully.")
        except Exception as e:
            logger.error(f"Error loading tokenizer: {e}")
            raise
        
    def analyze_emotions(self, text):
        """Analyze emotions using DistilBERT."""
        logger.info(f"Analyzing emotions for text: {text}")

        # Tokenize the input text
        inputs = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=512, padding=True)
        inputs = {k: v.to(self.device) for k, v in inputs.items()}

        # Perform inference
        with torch.no_grad():
            outputs = self.model(**inputs)
            logits = outputs.logits

        # Get predicted label
        predicted_label = torch.argmax(logits, dim=1).item()
        emotion = self.map_label_to_emotion(predicted_label)
        logger.info(f"Extracted emotion: {emotion}")

        return emotion, "Analysis complete."

    def map_label_to_emotion(self, label):
        """Map model output label to emotion string."""
        emotion_map = {
            0: "Negative",  # Adjust these based on your emotion mapping
            1: "Positive"
            # Extend the map based on your emotion requirements
        }
        return emotion_map.get(label, "Unknown")

# YouTube API setup
def get_youtube_service():
    """Build and return the YouTube service object."""
    return build('youtube', 'v3', developerKey=YOUTUBE_API_KEY)

def get_video_comments(video_id):
    """Fetch comments from a YouTube video using the video_id."""
    youtube = get_youtube_service()
    comments = []
                
    try:
        logger.info(f"Fetching comments for video_id: {video_id}")
        response = youtube.commentThreads().list(
            part='snippet',
            videoId=video_id,
            textFormat='plainText',
            maxResults=100
        ).execute()
        
        # Process the response to extract comments
        for item in response.get('items', []):
            comment = item['snippet']['topLevelComment']['snippet']['textDisplay']
            comments.append(comment)
            
        # Handle pagination if more comments are available
        while 'nextPageToken' in response:
            logger.info("Fetching next page of comments...")
            response = youtube.commentThreads().list(
                part='snippet',
                videoId=video_id,
                textFormat='plainText',
                maxResults=100,
                pageToken=response['nextPageToken']
            ).execute()

            for item in response.get('items', []):
                comment = item['snippet']['topLevelComment']['snippet']['textDisplay']
                comments.append(comment)

    except HttpError as e:
        logger.error(f"HTTP error fetching comments: {e}")
    except Exception as e:
        logger.error(f"Unexpected error fetching comments: {e}")
    
    logger.info(f"Total comments fetched: {len(comments)}")
    return comments
            
# Helper function to detect URLs in text
def contains_url(text):
    """Check if the text contains any URL."""
    url_pattern = re.compile(
        r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|'
        r'(?:%[0-9a-fA-F][0-9a-fA-F]))+'
    )
    return bool(url_pattern.search(text))
            
# Instantiate the EmotionAnalyzer
emotion_analyzer = EmotionAnalyzer()

@app.post("/analyze_emotion/", summary="Analyze emotions in YouTube video comments")
async def analyze_emotion(request: EmotionRequest):
    """     
    Analyze emotions for comments from a YouTube video.

    - **video_id**: The ID of the YouTube video to analyze
    """
    logger.info(f"Received request to analyze video_id: {request.video_id}")

    comments = get_video_comments(request.video_id)

    if not comments:
        logger.warning(f"No comments found for video_id: {request.video_id}")
        raise HTTPException(status_code=404, detail="No comments found for this video ID.")
    
    results = []
    for comment in comments:
        if contains_url(comment):
            # Mark comment as Neutral due to presence of URL
            emotion = "Neutral"
            generated_text = "Comment contains a link; marked as Neutral."
            logger.info(f"Comment contains URL. Marked as Neutral.")
        else:
            emotion, generated_text = emotion_analyzer.analyze_emotions(comment)
    
        results.append({
            "comment": comment,
            "emotion": emotion,
            "generated_text": generated_text
        })
    
    return results