from transformers import pipeline from collections import Counter import json from pathlib import Path from datetime import datetime import numpy as np class ChatSentimentAnalyzer: def __init__(self): # Initialize the emotion analysis pipeline with RoBERTa model self.analyzer = pipeline("text-classification", model='bhadresh-savani/roberta-base-emotion', return_all_scores=True) # Define emotion categories self.emotion_labels = ['anger', 'fear', 'joy', 'love', 'sadness', 'surprise'] # Map emotions to sentiment categories self.emotion_to_sentiment = { 'joy': 'POSITIVE', 'love': 'POSITIVE', 'surprise': 'NEUTRAL', 'anger': 'NEGATIVE', 'fear': 'NEGATIVE', 'sadness': 'NEGATIVE' } def analyze_chat_session(self, chat_data): """Analyze sentiment for a complete chat session""" if isinstance(chat_data, str): # If it's a file path, load the data with open(chat_data, 'r') as f: chat_data = json.load(f) messages = chat_data.get('messages', []) # Separate user and assistant messages user_messages = [] assistant_messages = [] for msg in messages: if msg['role'] == 'user': user_messages.append({ 'content': msg['content'], 'timestamp': msg['timestamp'] }) elif msg['role'] == 'assistant': assistant_messages.append({ 'content': msg['content'], 'timestamp': msg['timestamp'], 'agent': msg.get('agent', 'Unknown') }) # Analyze sentiments user_sentiments = self._analyze_messages(user_messages, 'user') assistant_sentiments = self._analyze_messages(assistant_messages, 'assistant') # Calculate session metrics session_metrics = self._calculate_session_metrics(user_sentiments, assistant_sentiments) return { 'user_sentiments': user_sentiments, 'assistant_sentiments': assistant_sentiments, 'session_metrics': session_metrics, 'session_id': chat_data.get('session_id'), 'user_name': chat_data.get('user_name'), 'timestamp': datetime.now().isoformat() } def _analyze_messages(self, messages, sender_type): """Analyze sentiment for a list of messages""" if not messages: return [] # Extract just the content for analysis contents = [msg['content'] for msg in messages] results = [] for msg, content in zip(messages, contents): try: # Get emotion predictions predictions = self.analyzer(content) # Find the emotion with highest score emotion_scores = predictions[0] # Get first (and only) result top_emotion = max(emotion_scores, key=lambda x: x['score']) # Create emotion distribution emotion_dist = {score['label']: round(score['score'], 3) for score in emotion_scores} # Map to sentiment sentiment = self.emotion_to_sentiment.get(top_emotion['label'], 'NEUTRAL') results.append({ 'content': msg['content'], 'timestamp': msg.get('timestamp'), 'emotion': top_emotion['label'], 'emotion_confidence': round(top_emotion['score'], 3), 'emotion_distribution': emotion_dist, 'sentiment': sentiment, 'sender_type': sender_type }) except Exception as e: print(f"Error analyzing message: {e}") results.append({ 'content': msg['content'], 'timestamp': msg.get('timestamp'), 'emotion': 'unknown', 'emotion_confidence': 0.0, 'emotion_distribution': {}, 'sentiment': 'NEUTRAL', 'sender_type': sender_type }) return results def analyze_limited_data(self, data): """Optimized analysis for limited data""" try: all_messages = [] session_count = len(data['sessions']) # Collect all messages for session in data['sessions']: all_messages.extend(session['messages']) if not all_messages: return {"error": "No messages to analyze"} # Limit total messages for performance all_messages = all_messages[:100] # Process emotions using the existing analyzer emotion_counts = {} positive_count = 0 negative_count = 0 for message in all_messages: try: # Get emotion predictions using existing analyzer predictions = self.analyzer(message) emotion_scores = predictions[0] top_emotion = max(emotion_scores, key=lambda x: x['score']) # Count emotions emotion = top_emotion['label'] emotion_counts[emotion] = emotion_counts.get(emotion, 0) + 1 # Map to sentiment using existing mapping sentiment = self.emotion_to_sentiment.get(emotion, 'NEUTRAL') if sentiment == 'POSITIVE': positive_count += 1 elif sentiment == 'NEGATIVE': negative_count += 1 except Exception as e: print(f"Error processing message: {e}") continue # Calculate quick metrics total_messages = len(all_messages) positivity_ratio = positive_count / total_messages if total_messages > 0 else 0 # Determine dominant emotion dominant_emotion = max(emotion_counts.items(), key=lambda x: x[1])[0] if emotion_counts else 'unknown' # Simple trend calculation if positive_count > negative_count * 1.5: trend = 'IMPROVING' elif negative_count > positive_count * 1.5: trend = 'DECLINING' else: trend = 'STABLE' return { 'total_sessions': session_count, 'total_messages_analyzed': total_messages, 'aggregated_metrics': { 'overall_positivity_ratio': positivity_ratio, 'dominant_emotion': dominant_emotion, 'emotion_distribution': emotion_counts, 'total_positive_messages': positive_count, 'total_negative_messages': negative_count, 'most_common_trend': trend, 'trend_distribution': { 'IMPROVING': 1 if trend == 'IMPROVING' else 0, 'STABLE': 1 if trend == 'STABLE' else 0, 'DECLINING': 1 if trend == 'DECLINING' else 0 } } } except Exception as e: print(f"Error in limited analysis: {e}") return {"error": str(e)} def _calculate_session_metrics(self, user_sentiments, assistant_sentiments): """Calculate overall session metrics""" # Extract emotions and sentiments user_emotions = [s['emotion'] for s in user_sentiments if s['emotion'] != 'unknown'] user_sentiment_labels = [s['sentiment'] for s in user_sentiments] metrics = { 'user_emotion_distribution': Counter(user_emotions), 'user_sentiment_distribution': Counter(user_sentiment_labels), 'assistant_sentiment_distribution': Counter([s['sentiment'] for s in assistant_sentiments]), 'user_avg_confidence': np.mean([s['emotion_confidence'] for s in user_sentiments]) if user_sentiments else 0, 'assistant_avg_confidence': np.mean([s['emotion_confidence'] for s in assistant_sentiments]) if assistant_sentiments else 0, 'total_messages': len(user_sentiments) + len(assistant_sentiments), 'sentiment_trend': self._calculate_sentiment_trend(user_sentiments), 'dominant_emotion': Counter(user_emotions).most_common(1)[0][0] if user_emotions else 'unknown' } # Calculate emotional valence if user_emotions: positive_emotions = sum(1 for e in user_emotions if e in ['joy', 'love']) negative_emotions = sum(1 for e in user_emotions if e in ['anger', 'fear', 'sadness']) neutral_emotions = sum(1 for e in user_emotions if e in ['surprise']) total_emotions = len(user_emotions) metrics['emotional_valence'] = { 'positive': round(positive_emotions / total_emotions, 3), 'negative': round(negative_emotions / total_emotions, 3), 'neutral': round(neutral_emotions / total_emotions, 3) } # Calculate overall session sentiment if user_sentiment_labels: positive_count = user_sentiment_labels.count('POSITIVE') negative_count = user_sentiment_labels.count('NEGATIVE') metrics['overall_session_sentiment'] = 'POSITIVE' if positive_count > negative_count else 'NEGATIVE' metrics['positivity_ratio'] = positive_count / len(user_sentiment_labels) if user_sentiment_labels else 0 return metrics def _calculate_sentiment_trend(self, sentiments): """Calculate how sentiment changes over time""" if len(sentiments) < 2: return 'stable' # Convert sentiments to numeric values based on emotional valence sentiment_values = [] for s in sentiments: if s['emotion'] in ['joy', 'love']: sentiment_values.append(1) elif s['emotion'] in ['anger', 'fear', 'sadness']: sentiment_values.append(-1) else: sentiment_values.append(0) # Calculate trend using simple linear regression x = np.arange(len(sentiment_values)) y = np.array(sentiment_values) # Calculate slope if len(x) > 1: slope = np.polyfit(x, y, 1)[0] if slope > 0.1: return 'improving' elif slope < -0.1: return 'declining' else: return 'stable' return 'stable' def analyze_user_history(self, user_name, sessions_dir='chat_sessions'): """Analyze all chat sessions for a specific user""" sessions_path = Path(sessions_dir) user_files = list(sessions_path.glob(f"chat_{user_name}_*.json")) all_analyses = [] for file_path in sorted(user_files): try: analysis = self.analyze_chat_session(str(file_path)) all_analyses.append(analysis) except Exception as e: print(f"Error analyzing {file_path}: {e}") # Aggregate metrics across all sessions if all_analyses: aggregated_metrics = self._aggregate_user_metrics(all_analyses) return { 'user_name': user_name, 'total_sessions': len(all_analyses), 'session_analyses': all_analyses, 'aggregated_metrics': aggregated_metrics } return None def _aggregate_user_metrics(self, analyses): """Aggregate metrics across multiple sessions""" total_positive = 0 total_negative = 0 total_neutral = 0 all_trends = [] all_emotions = [] for analysis in analyses: metrics = analysis['session_metrics'] user_dist = metrics['user_sentiment_distribution'] emotion_dist = metrics['user_emotion_distribution'] total_positive += user_dist.get('POSITIVE', 0) total_negative += user_dist.get('NEGATIVE', 0) total_neutral += user_dist.get('NEUTRAL', 0) all_trends.append(metrics['sentiment_trend']) # Collect all emotions for emotion, count in emotion_dist.items(): all_emotions.extend([emotion] * count) trend_counts = Counter(all_trends) emotion_counts = Counter(all_emotions) return { 'total_positive_messages': total_positive, 'total_negative_messages': total_negative, 'total_neutral_messages': total_neutral, 'overall_positivity_ratio': total_positive / (total_positive + total_negative + total_neutral) if (total_positive + total_negative + total_neutral) > 0 else 0, 'trend_distribution': dict(trend_counts), 'most_common_trend': trend_counts.most_common(1)[0][0] if trend_counts else 'unknown', 'emotion_distribution': dict(emotion_counts), 'dominant_emotion': emotion_counts.most_common(1)[0][0] if emotion_counts else 'unknown' } # Backward compatibility wrapper def analyze_sentiments(messages, label): analyzer = ChatSentimentAnalyzer() results = [] for msg in messages: try: predictions = analyzer.analyzer(msg) emotion_scores = predictions[0] top_emotion = max(emotion_scores, key=lambda x: x['score']) results.append({ 'label': top_emotion['label'], 'score': top_emotion['score'] }) print(f"Message: {msg}") print(f"Emotion: {top_emotion['label']} (Score: {round(top_emotion['score'], 2)})") print(f"All emotions: {[(e['label'], round(e['score'], 3)) for e in emotion_scores]}\n") except: pass return results def summarize_sentiments(sentiments, label): labels = [r["label"] for r in sentiments] summary = Counter(labels) print(f"📊 {label} Emotion Summary: {dict(summary)}\n") return summary # Example usage if __name__ == "__main__": analyzer = ChatSentimentAnalyzer() # Test the emotion classifier test_messages = [ "I love using transformers. The best part is wide range of support and its easy to use", "I am feeling very anxious about the upcoming exam", "This makes me so angry and frustrated", "I'm surprised by how well this works" ] print("Testing emotion analysis:") print("-" * 50) for msg in test_messages: predictions = analyzer.analyzer(msg) emotion_scores = predictions[0] top_emotion = max(emotion_scores, key=lambda x: x['score']) print(f"Message: {msg}") print(f"Top emotion: {top_emotion['label']} ({round(top_emotion['score'], 3)})") print(f"All emotions: {[(e['label'], round(e['score'], 3)) for e in emotion_scores]}") print("-" * 50)