""" Sentiment Analyzer Service - Analyzes sentiment of news articles using TextBlob and VADER """ from textblob import TextBlob from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer from typing import Dict, Tuple import logging from ..config import settings from ..utils.logger import setup_logger logger = setup_logger(__name__) class SentimentAnalyzer: """ Hybrid sentiment analyzer using both TextBlob and VADER. """ def __init__(self): """Initialize sentiment analyzers.""" self.vader = SentimentIntensityAnalyzer() self.threshold_negative = settings.SENTIMENT_THRESHOLD_NEGATIVE self.threshold_positive = settings.SENTIMENT_THRESHOLD_POSITIVE def analyze_text(self, text: str, title: str = "") -> Dict[str, any]: """ Analyze sentiment of text using hybrid approach. Args: text: Main text to analyze (article description/summary) title: Optional title (weighted more heavily) Returns: Dictionary with: - score: float (-1 to 1) - label: str ('positive', 'neutral', 'negative') - analyzer: str ('hybrid') - textblob_score: float - vader_score: float """ if not text or not text.strip(): return { 'score': 0.0, 'label': 'neutral', 'analyzer': 'hybrid', 'textblob_score': 0.0, 'vader_score': 0.0 } try: # Combine title and text, giving title more weight if title: combined_text = f"{title} {title} {text}" # Title appears twice for emphasis else: combined_text = text # TextBlob analysis textblob_score = self._analyze_textblob(combined_text) # VADER analysis vader_score = self._analyze_vader(combined_text) # Combine scores (weighted average: 40% TextBlob, 60% VADER) # VADER is better for news/social media combined_score = (0.4 * textblob_score) + (0.6 * vader_score) # Get sentiment label label = self.get_sentiment_label(combined_score) # Override for crisis keywords - prevent positive labeling of negative events crisis_keywords = [ 'attack', 'strike', 'kill', 'death', 'dead', 'war', 'crisis', 'disaster', 'earthquake', 'tsunami', 'terror', 'shooting', 'explosion', 'emergency', 'critical', 'urgent', 'bombing', 'violence', 'conflict', 'casualties', 'victim', 'threat' ] combined_text_lower = f"{title} {text}".lower() has_crisis = any(keyword in combined_text_lower for keyword in crisis_keywords) if has_crisis and combined_score > 0: # Force crisis news to neutral or negative, never positive combined_score = min(combined_score, 0.0) label = self.get_sentiment_label(combined_score) logger.debug(f"Crisis keyword detected - overriding to {label} (score: {combined_score:.3f})") logger.debug(f"Sentiment analysis - TextBlob: {textblob_score:.3f}, VADER: {vader_score:.3f}, Combined: {combined_score:.3f}, Label: {label}") return { 'score': round(combined_score, 4), 'label': label, 'analyzer': 'hybrid', 'textblob_score': round(textblob_score, 4), 'vader_score': round(vader_score, 4) } except Exception as e: logger.error(f"Error analyzing sentiment: {e}") return { 'score': 0.0, 'label': 'neutral', 'analyzer': 'hybrid', 'textblob_score': 0.0, 'vader_score': 0.0 } def _analyze_textblob(self, text: str) -> float: """ Analyze sentiment using TextBlob. Args: text: Text to analyze Returns: Polarity score from -1 to 1 """ try: blob = TextBlob(text) return blob.sentiment.polarity except Exception as e: logger.warning(f"TextBlob analysis failed: {e}") return 0.0 def _analyze_vader(self, text: str) -> float: """ Analyze sentiment using VADER. Args: text: Text to analyze Returns: Compound score from -1 to 1 """ try: scores = self.vader.polarity_scores(text) return scores['compound'] except Exception as e: logger.warning(f"VADER analysis failed: {e}") return 0.0 def get_sentiment_label(self, score: float) -> str: """ Convert sentiment score to label. Args: score: Sentiment score from -1 to 1 Returns: 'positive', 'neutral', or 'negative' """ if score >= self.threshold_positive: return 'positive' elif score <= self.threshold_negative: return 'negative' else: return 'neutral' def analyze_batch(self, articles: list) -> list: """ Analyze sentiment for a batch of articles. Args: articles: List of article dictionaries with 'title' and 'summary' Returns: List of articles with added sentiment fields """ logger.info(f"Analyzing sentiment for {len(articles)} articles") for article in articles: title = article.get('title', '') summary = article.get('summary', '') sentiment = self.analyze_text(summary, title) article['sentiment_score'] = sentiment['score'] article['sentiment_label'] = sentiment['label'] article['sentiment_analyzer'] = sentiment['analyzer'] logger.info(f"Sentiment analysis complete for {len(articles)} articles") return articles # Example usage if __name__ == "__main__": analyzer = SentimentAnalyzer() # Test cases test_cases = [ { 'title': 'Major Earthquake Hits City', 'text': 'A devastating earthquake has struck the city, causing widespread destruction and casualties.' }, { 'title': 'Economy Shows Strong Growth', 'text': 'The economy is booming with record job creation and rising consumer confidence.' }, { 'title': 'Weather Update', 'text': 'Partly cloudy skies expected today with temperatures in the mid-70s.' } ] for case in test_cases: result = analyzer.analyze_text(case['text'], case['title']) print(f"Title: {case['title']}") print(f"Score: {result['score']}, Label: {result['label']}") print(f"TextBlob: {result['textblob_score']}, VADER: {result['vader_score']}") print("---")