import pandas as pd
import numpy as np
from transformers import AutoTokenizer, AutoModel
import torch
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from textblob import TextBlob
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.decomposition import PCA
import warnings
from typing import Dict, List, Tuple
import logging
from collections import Counter
from detoxify import Detoxify
import re
from datetime import datetime
import seaborn as sns
import matplotlib.pyplot as plt
from pathlib import Path
import json
class AdvancedYelpAnalyzer:
def __init__(self, df: pd.DataFrame):
"""Initialize the analyzer with necessary models and configurations"""
self.df = df.copy()
self.bert_tokenizer = AutoTokenizer.from_pretrained('bert-base-uncased')
self.bert_model = AutoModel.from_pretrained('bert-base-uncased')
self.vader = SentimentIntensityAnalyzer()
self.toxic_model = Detoxify('original')
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
self.bert_model.to(self.device)
# Configure logging
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
def get_bert_embeddings(self, texts: List[str], batch_size: int = 32) -> np.ndarray:
"""Generate BERT embeddings for text"""
embeddings = []
for i in range(0, len(texts), batch_size):
batch_texts = texts[i:i + batch_size]
encoded = self.bert_tokenizer(batch_texts,
padding=True,
truncation=True,
max_length=512,
return_tensors='pt')
with torch.no_grad():
encoded = {k: v.to(self.device) for k, v in encoded.items()}
outputs = self.bert_model(**encoded)
batch_embeddings = outputs.last_hidden_state[:, 0, :].cpu().numpy()
embeddings.append(batch_embeddings)
return np.vstack(embeddings)
def analyze_sentiment(self) -> pd.DataFrame:
"""Perform comprehensive sentiment analysis using multiple tools"""
self.logger.info("Starting sentiment analysis...")
# Calculate BERT embeddings for reviews
self.logger.info("Calculating BERT embeddings...")
review_texts = self.df['review_text'].fillna('').tolist()
bert_embeddings = self.get_bert_embeddings(review_texts)
# Calculate review length using BERT tokenizer
self.logger.info("Calculating tokenized lengths...")
self.df['review_length'] = self.df['review_text'].apply(
lambda x: len(self.bert_tokenizer.encode(str(x)))
)
# Store BERT embeddings mean and std as features
self.df['bert_embedding_mean'] = np.mean(bert_embeddings, axis=1)
self.df['bert_embedding_std'] = np.std(bert_embeddings, axis=1)
# TextBlob sentiment and subjectivity
self.df['textblob_polarity'] = self.df['review_text'].apply(
lambda x: TextBlob(str(x)).sentiment.polarity
)
self.df['textblob_subjectivity'] = self.df['review_text'].apply(
lambda x: TextBlob(str(x)).sentiment.subjectivity
)
# VADER sentiment with custom negative phrase handling
def get_enhanced_vader_scores(text):
# Custom negative phrases
negative_phrases = [
'too long', 'way too long', 'waiting', 'changed our minds',
'too many', 'took forever', 'took too long', 'waste of time',
'not worth', 'disappointing', 'mediocre', 'needs improvement'
]
# Get base VADER scores
base_scores = self.vader.polarity_scores(str(text))
# Check for negative phrases
text_lower = str(text).lower()
neg_count = sum(1 for phrase in negative_phrases if phrase in text_lower)
# Adjust scores if negative phrases are found
if neg_count > 0:
base_scores['neg'] = max(base_scores['neg'], min(0.7, neg_count * 0.2))
base_scores['compound'] *= (1 - (neg_count * 0.15))
# Readjust neutral score
base_scores['neu'] = max(0, 1 - base_scores['neg'] - base_scores['pos'])
return base_scores
# Apply enhanced VADER scoring
vader_scores = self.df['review_text'].apply(get_enhanced_vader_scores)
self.df['vader_compound'] = vader_scores.apply(lambda x: x['compound'])
self.df['vader_negative'] = vader_scores.apply(lambda x: x['neg'])
self.df['vader_positive'] = vader_scores.apply(lambda x: x['pos'])
self.df['vader_neutral'] = vader_scores.apply(lambda x: x['neu'])
# Calculate sentiment extremity
self.df['sentiment_extremity'] = self.df['vader_compound'].abs()
return self.df
def detect_anomalies(self) -> pd.DataFrame:
"""Detect anomalous reviews using Isolation Forest with BERT features"""
self.logger.info("Detecting anomalies...")
# Prepare features for anomaly detection
features = [
'review_stars',
'textblob_polarity',
'vader_compound',
'sentiment_extremity',
'review_length',
'bert_embedding_mean',
'bert_embedding_std'
]
# Ensure all features exist
missing_features = [f for f in features if f not in self.df.columns]
if missing_features:
self.analyze_sentiment()
# Standardize features
scaler = StandardScaler()
X = scaler.fit_transform(self.df[features])
# Apply Isolation Forest
iso_forest = IsolationForest(
contamination=0.1,
random_state=42,
n_jobs=-1
)
# Fit and predict
self.df['is_anomaly'] = iso_forest.fit_predict(X)
self.df['anomaly_score'] = iso_forest.score_samples(X)
return self.df
def detect_ai_generated_text(self) -> pd.DataFrame:
"""Estimate likelihood of AI-generated content"""
self.logger.info("Detecting AI-generated content...")
# Ensure sentiment analysis has been run
if 'textblob_subjectivity' not in self.df.columns:
self.analyze_sentiment()
# Use detoxify model to get toxicity scores
texts = self.df['review_text'].fillna('').tolist()
toxic_scores = self.toxic_model.predict(texts)
# Add scores to DataFrame
toxic_score_types = ['toxicity', 'severe_toxicity', 'obscene', 'identity_attack',
'insult', 'threat', 'sexual_explicit']
for score_type in toxic_score_types:
if score_type in toxic_scores:
self.df[f'toxic_{score_type}'] = toxic_scores[score_type]
# Calculate AI generation likelihood based on various factors
self.df['ai_generated_likelihood'] = (
(self.df['textblob_subjectivity'] < 0.3) & # Low subjectivity
(self.df['sentiment_extremity'] > 0.8) & # Extreme sentiment
(self.df['review_length'] > self.df['review_length'].quantile(0.95)) & # Unusually long
(self.df['bert_embedding_std'] < self.df['bert_embedding_std'].quantile(0.25)) # Unusual language patterns
).astype(int)
# Add additional AI detection features
self.df['ai_detection_score'] = (
(self.df['textblob_subjectivity'] * -1) + # Lower subjectivity increases score
(self.df['sentiment_extremity'] * 0.5) + # Extreme sentiment contributes somewhat
(self.df['bert_embedding_std'] * -0.5) # Lower variation in embeddings increases score
).clip(0, 1) # Normalize between 0 and 1
return self.df
def analyze_business_categories(self) -> Dict:
"""Analyze trends and patterns specific to business categories"""
self.logger.info("Analyzing business categories...")
# Extract categories
categories = self.df['categories'].fillna('').str.split(', ')
all_categories = [cat for cats in categories if isinstance(cats, list) for cat in cats]
category_counts = Counter(all_categories)
# Analyze reviews by category
category_analysis = {}
for category in set(all_categories):
category_reviews = self.df[self.df['categories'].str.contains(category, na=False)]
category_analysis[category] = {
'review_count': len(category_reviews),
'avg_rating': category_reviews['review_stars'].mean() if not category_reviews.empty else None,
'avg_sentiment': category_reviews['vader_compound'].mean() if 'vader_compound' in self.df.columns and not category_reviews.empty else None,
'avg_subjectivity': category_reviews['textblob_subjectivity'].mean() if 'textblob_subjectivity' in self.df.columns and not category_reviews.empty else None
}
return category_analysis
def visualize_results(self, output_dir: str):
"""Create visualizations for analysis results"""
plt.figure(figsize=(15, 10))
# Sentiment Distribution
plt.subplot(2, 2, 1)
sns.histplot(data=self.df, x='vader_compound', bins=50)
plt.title('Sentiment Distribution')
# Review Volume Over Time
plt.subplot(2, 2, 2)
daily_reviews = self.df.groupby('review_date').size()
daily_reviews.plot()
plt.title('Review Volume Over Time')
# Anomaly Score Distribution
plt.subplot(2, 2, 3)
if 'anomaly_score' not in self.df.columns:
self.detect_anomalies()
sns.histplot(data=self.df, x='anomaly_score', bins=50)
plt.title('Anomaly Score Distribution')
# AI Generation Likelihood
plt.subplot(2, 2, 4)
if 'ai_generated_likelihood' not in self.df.columns:
self.detect_ai_generated_text()
sns.histplot(data=self.df, x='ai_generated_likelihood', bins=2)
plt.title('AI Generation Likelihood')
plt.tight_layout()
plt.savefig(f'{output_dir}/analysis_results.png')
plt.close()
def run_full_analysis(self, output_dir: str) -> Tuple[pd.DataFrame, Dict]:
"""Run complete analysis pipeline with detailed outputs"""
self.logger.info("Starting full analysis pipeline...")
# Create output directory if it doesn't exist
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
try:
# Run all analyses
self.analyze_sentiment()
self.detect_anomalies()
self.detect_ai_generated_text()
category_analysis = self.analyze_business_categories()
# Create visualizations
self.visualize_results(str(output_dir))
# Compile results
analysis_results = {
'category_analysis': category_analysis,
'sentiment_summary': {
'avg_sentiment': self.df['vader_compound'].mean(),
'positive_reviews': len(self.df[self.df['vader_compound'] > 0.5]),
'negative_reviews': len(self.df[self.df['vader_compound'] < -0.5]),
'neutral_reviews': len(self.df[abs(self.df['vader_compound']) <= 0.5])
},
'ai_detection_summary': {
'likely_ai_generated': len(self.df[self.df['ai_generated_likelihood'] == 1]),
'avg_ai_score': self.df['ai_detection_score'].mean()
},
'anomaly_summary': {
'anomalous_reviews': len(self.df[self.df['is_anomaly'] == -1]),
'avg_anomaly_score': self.df['anomaly_score'].mean()
}
}
# Save results
self.df.to_csv(output_dir / "analyzed_data.csv", index=False)
with open(output_dir / "analysis_results.json", 'w') as f:
json.dump(analysis_results, f, indent=4)
return self.df, analysis_results
except Exception as e:
self.logger.error(f"Error during analysis: {str(e)}")
raise
# For testing
if __name__ == "__main__":
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
try:
# Read test data
df = pd.read_csv("test_data.csv")
# Initialize analyzer
analyzer = AdvancedYelpAnalyzer(df)
# Run analysis
output_dir = "output"
analyzed_df, results = analyzer.run_full_analysis(output_dir)
logger.info("Analysis completed successfully!")
except Exception as e:
logger.error(f"Error during testing: {str(e)}")
raise