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