📝 INTRODUCTION

This project implements a machine learning-based sentiment analyzer that can classify movie or product reviews into Positive, Negative, or Neutral categories. It uses multiple algorithms including Naive Bayes, Logistic Regression, and LSTM (deep learning) for comparison. The system includes comprehensive text preprocessing, feature extraction, and model evaluation. MongoDB is used to store reviews, predictions, and model performance metrics.

✨ FEATURES

1. Multi-class Sentiment Analysis: Classify reviews as Positive, Negative, or Neutral
2. Multiple Algorithms: Compare performance of different ML/DL models
3. Real-time Analysis: Analyze individual reviews instantly
4. Batch Processing: Analyze multiple reviews at once
5. Model Training & Evaluation: Train and compare multiple models
6. Visualization: Generate sentiment distribution charts and confusion matrices
7. Export Functionality: Export results to CSV/JSON/Excel
8. API Endpoints: RESTful API for integration with other applications
9. Interactive Dashboard: Web-based interface for easy interaction
10. Performance Metrics: Track accuracy, precision, recall, and F1-score

📁 PROJECT STRUCTURE

sentiment-analyzer/
│
├── config/
│   ├── mongodb_config.py
│   └── model_config.py
│
├── models/
│   ├── sentiment_classifier.py
│   ├── deep_learning_model.py
│   └── model_comparator.py
│
├── database/
│   ├── db_operations.py
│   └── review_schema.py
│
├── utils/
│   ├── text_preprocessing.py
│   ├── feature_extraction.py
│   └── visualization.py
│
├── api/
│   ├── app.py
│   └── routes.py
│
├── web/
│   ├── templates/
│   │   ├── index.html
│   │   ├── analyze.html
│   │   └── dashboard.html
│   └── static/
│       ├── css/
│       │   └── style.css
│       └── js/
│           └── main.js
│
├── data/
│   ├── movie_reviews.csv
│   └── product_reviews.csv
│
├── requirements.txt
├── .env
└── README.md

🚀 COMPLETE CODE

1. requirements.txt

pymongo==4.5.0
scikit-learn==1.3.0
pandas==2.0.3
numpy==1.24.3
nltk==3.8.1
tensorflow==2.13.0
keras==2.13.1
transformers==4.35.0
joblib==1.3.2
python-dotenv==1.0.0
flask==2.3.2
flask-cors==4.0.0
flask-restful==0.3.10
matplotlib==3.7.2
seaborn==0.12.2
plotly==5.17.0
wordcloud==1.9.2
textblob==0.17.1
vaderSentiment==3.3.2
gunicorn==21.2.0
celery==5.3.4
redis==5.0.1

2. config/mongodb_config.py

import os
from dotenv import load_dotenv
load_dotenv()
class MongoDBConfig:
# MongoDB connection settings
MONGODB_URI = os.getenv('MONGODB_URI', 'mongodb://localhost:27017/')
DATABASE_NAME = os.getenv('DATABASE_NAME', 'sentiment_analyzer_db')
# Collections
REVIEWS_COLLECTION = 'reviews'
PREDICTIONS_COLLECTION = 'predictions'
MODEL_METRICS_COLLECTION = 'model_metrics'
TRAINING_DATA_COLLECTION = 'training_data'
# Connection settings
MAX_POOL_SIZE = 100
MIN_POOL_SIZE = 10
MAX_IDLE_TIME_MS = 10000
RETRY_WRITES = True

3. config/model_config.py

class ModelConfig:
# Model types
AVAILABLE_MODELS = ['naive_bayes', 'logistic_regression', 'svm', 'lstm', 'bert']
# Default model
DEFAULT_MODEL = 'logistic_regression'
# Text preprocessing settings
MAX_FEATURES = 5000
MAX_SEQUENCE_LENGTH = 200
EMBEDDING_DIM = 100
# Deep learning settings
BATCH_SIZE = 32
EPOCHS = 10
VALIDATION_SPLIT = 0.2
# Sentiment labels
SENTIMENT_LABELS = {
0: 'Negative',
1: 'Neutral',
2: 'Positive'
}
# Label mapping
LABEL_MAPPING = {
'negative': 0,
'neutral': 1,
'positive': 2
}

4. database/review_schema.py

from datetime import datetime
from bson import ObjectId
class ReviewSchema:
"""Schema for review documents in MongoDB"""
@staticmethod
def get_review_schema(review_text, source='user', metadata=None):
"""Create a review document following the schema"""
return {
'review_text': review_text,
'source': source,
'sentiment_label': None,
'sentiment_score': None,
'confidence': None,
'created_at': datetime.utcnow(),
'processed_at': None,
'review_length': len(review_text),
'word_count': len(review_text.split()),
'metadata': metadata or {
'has_emojis': False,
'has_uppercase': any(c.isupper() for c in review_text),
'has_punctuation': any(c in '!?.' for c in review_text)
}
}
@staticmethod
def get_prediction_schema(review_id, review_text, prediction, confidence, 
actual_label=None, model_used='default'):
"""Create a prediction document"""
return {
'review_id': ObjectId(review_id) if isinstance(review_id, str) else review_id,
'review_text': review_text,
'prediction': prediction,
'confidence': confidence,
'actual_label': actual_label,
'is_correct': prediction == actual_label if actual_label else None,
'model_used': model_used,
'created_at': datetime.utcnow()
}
@staticmethod
def get_metrics_schema(metrics):
"""Create model metrics document"""
return {
'model_name': metrics['model_name'],
'accuracy': metrics['accuracy'],
'precision': metrics['precision'],
'recall': metrics['recall'],
'f1_score': metrics['f1_score'],
'confusion_matrix': metrics['confusion_matrix'],
'classification_report': metrics.get('classification_report', {}),
'training_time': metrics.get('training_time', 0),
'created_at': datetime.utcnow()
}

5. database/db_operations.py

from pymongo import MongoClient, errors
from pymongo.errors import ConnectionFailure
from config.mongodb_config import MongoDBConfig
from database.review_schema import ReviewSchema
import logging
from datetime import datetime, timedelta
import pandas as pd
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MongoDBOperations:
def __init__(self):
self.config = MongoDBConfig()
self.client = None
self.db = None
self.connect()
def connect(self):
"""Establish connection to MongoDB"""
try:
self.client = MongoClient(
self.config.MONGODB_URI,
maxPoolSize=self.config.MAX_POOL_SIZE,
minPoolSize=self.config.MIN_POOL_SIZE,
maxIdleTimeMS=self.config.MAX_IDLE_TIME_MS,
retryWrites=self.config.RETRY_WRITES
)
self.db = self.client[self.config.DATABASE_NAME]
# Test connection
self.client.admin.command('ping')
logger.info("Successfully connected to MongoDB")
# Create indexes
self.create_indexes()
except ConnectionFailure as e:
logger.error(f"Failed to connect to MongoDB: {e}")
raise
def create_indexes(self):
"""Create necessary indexes for better query performance"""
try:
# Reviews collection indexes
self.db[self.config.REVIEWS_COLLECTION].create_index('created_at')
self.db[self.config.REVIEWS_COLLECTION].create_index('sentiment_label')
self.db[self.config.REVIEWS_COLLECTION].create_index('source')
# Predictions collection indexes
self.db[self.config.PREDICTIONS_COLLECTION].create_index('created_at')
self.db[self.config.PREDICTIONS_COLLECTION].create_index('prediction')
self.db[self.config.PREDICTIONS_COLLECTION].create_index('model_used')
# Model metrics collection indexes
self.db[self.config.MODEL_METRICS_COLLECTION].create_index('model_name')
self.db[self.config.MODEL_METRICS_COLLECTION].create_index('created_at')
logger.info("Database indexes created successfully")
except Exception as e:
logger.error(f"Error creating indexes: {e}")
def insert_review(self, review_text, source='user', metadata=None):
"""Insert a single review into database"""
try:
review_doc = ReviewSchema.get_review_schema(review_text, source, metadata)
result = self.db[self.config.REVIEWS_COLLECTION].insert_one(review_doc)
# Update processed_at
self.db[self.config.REVIEWS_COLLECTION].update_one(
{'_id': result.inserted_id},
{'$set': {'processed_at': datetime.utcnow()}}
)
logger.info(f"Review inserted with ID: {result.inserted_id}")
return result.inserted_id
except Exception as e:
logger.error(f"Error inserting review: {e}")
return None
def insert_many_reviews(self, reviews_list, source='bulk_upload'):
"""Insert multiple reviews"""
try:
review_docs = [
ReviewSchema.get_review_schema(review, source) 
for review in reviews_list
]
result = self.db[self.config.REVIEWS_COLLECTION].insert_many(review_docs)
logger.info(f"Inserted {len(result.inserted_ids)} reviews")
return result.inserted_ids
except Exception as e:
logger.error(f"Error inserting multiple reviews: {e}")
return None
def save_prediction(self, review_id, review_text, prediction, confidence, 
actual_label=None, model_used='default'):
"""Save a prediction result"""
try:
prediction_doc = ReviewSchema.get_prediction_schema(
review_id, review_text, prediction, confidence, 
actual_label, model_used
)
result = self.db[self.config.PREDICTIONS_COLLECTION].insert_one(prediction_doc)
# Update the review with prediction
self.db[self.config.REVIEWS_COLLECTION].update_one(
{'_id': ObjectId(review_id) if isinstance(review_id, str) else review_id},
{
'$set': {
'sentiment_label': prediction,
'sentiment_score': confidence,
'confidence': confidence
}
}
)
return result.inserted_id
except Exception as e:
logger.error(f"Error saving prediction: {e}")
return None
def update_review_label(self, review_id, label, confidence=None):
"""Update review with actual sentiment label"""
try:
update_data = {'sentiment_label': label}
if confidence:
update_data['confidence'] = confidence
result = self.db[self.config.REVIEWS_COLLECTION].update_one(
{'_id': ObjectId(review_id) if isinstance(review_id, str) else review_id},
{'$set': update_data}
)
return result.modified_count > 0
except Exception as e:
logger.error(f"Error updating review label: {e}")
return False
def get_training_data(self, limit=None, labeled_only=True):
"""Retrieve training data from database"""
try:
query = {'sentiment_label': {'$ne': None}} if labeled_only else {}
cursor = self.db[self.config.REVIEWS_COLLECTION].find(query)
if limit:
cursor = cursor.limit(limit)
reviews = []
labels = []
for doc in cursor:
reviews.append(doc['review_text'])
labels.append(doc['sentiment_label'])
return reviews, labels
except Exception as e:
logger.error(f"Error retrieving training data: {e}")
return [], []
def get_reviews_by_sentiment(self, sentiment):
"""Get all reviews with specific sentiment"""
try:
cursor = self.db[self.config.REVIEWS_COLLECTION].find(
{'sentiment_label': sentiment}
).sort('created_at', -1)
return list(cursor)
except Exception as e:
logger.error(f"Error retrieving reviews by sentiment: {e}")
return []
def get_prediction_statistics(self, days=30):
"""Get statistics about predictions for the last N days"""
try:
cutoff_date = datetime.utcnow() - timedelta(days=days)
pipeline = [
{
'$match': {
'created_at': {'$gte': cutoff_date}
}
},
{
'$group': {
'_id': {
'prediction': '$prediction',
'model_used': '$model_used'
},
'count': {'$sum': 1},
'avg_confidence': {'$avg': '$confidence'}
}
},
{
'$sort': {'count': -1}
}
]
stats = list(self.db[self.config.PREDICTIONS_COLLECTION].aggregate(pipeline))
return stats
except Exception as e:
logger.error(f"Error getting prediction statistics: {e}")
return []
def get_model_performance(self, model_name=None):
"""Get model performance metrics"""
try:
query = {}
if model_name:
query['model_name'] = model_name
cursor = self.db[self.config.MODEL_METRICS_COLLECTION].find(
query
).sort('created_at', -1).limit(10)
return list(cursor)
except Exception as e:
logger.error(f"Error getting model performance: {e}")
return []
def export_reviews_to_dataframe(self):
"""Export all reviews to pandas DataFrame"""
try:
cursor = self.db[self.config.REVIEWS_COLLECTION].find()
df = pd.DataFrame(list(cursor))
# Convert ObjectId to string
if '_id' in df.columns:
df['_id'] = df['_id'].astype(str)
return df
except Exception as e:
logger.error(f"Error exporting to DataFrame: {e}")
return pd.DataFrame()
def close_connection(self):
"""Close MongoDB connection"""
if self.client:
self.client.close()
logger.info("MongoDB connection closed")

6. utils/text_preprocessing.py

import re
import nltk
from nltk.corpus import stopwords
from nltk.stem import WordNetLemmatizer
from nltk.tokenize import word_tokenize
import emoji
import string
# Download required NLTK data
try:
nltk.data.find('tokenizers/punkt')
nltk.data.find('corpora/stopwords')
nltk.data.find('corpora/wordnet')
except LookupError:
nltk.download('punkt')
nltk.download('stopwords')
nltk.download('wordnet')
nltk.download('averaged_perceptron_tagger')
class TextPreprocessor:
def __init__(self):
self.lemmatizer = WordNetLemmatizer()
self.stop_words = set(stopwords.words('english'))
def clean_text(self, text):
"""Clean and preprocess text"""
if not isinstance(text, str):
text = str(text)
# Convert to lowercase
text = text.lower()
# Remove HTML tags
text = re.sub(r'<.*?>', '', text)
# Remove URLs
text = re.sub(r'http\S+|www\S+|https\S+', '', text, flags=re.MULTILINE)
# Remove email addresses
text = re.sub(r'\S+@\S+', '', text)
# Remove numbers
text = re.sub(r'\d+', '', text)
# Remove punctuation
text = text.translate(str.maketrans('', '', string.punctuation))
# Remove extra whitespace
text = re.sub(r'\s+', ' ', text).strip()
return text
def convert_emojis(self, text):
"""Convert emojis to text representation"""
return emoji.demojize(text)
def tokenize(self, text):
"""Tokenize text into words"""
return word_tokenize(text)
def remove_stopwords(self, tokens):
"""Remove stopwords from tokens"""
return [token for token in tokens if token not in self.stop_words]
def lemmatize(self, tokens):
"""Apply lemmatization to tokens"""
return [self.lemmatizer.lemmatize(token) for token in tokens]
def get_pos_tags(self, tokens):
"""Get part-of-speech tags"""
return nltk.pos_tag(tokens)
def extract_sentiment_features(self, text):
"""Extract features relevant for sentiment analysis"""
features = {}
# Count of exclamation marks
features['exclamation_count'] = text.count('!')
# Count of question marks
features['question_count'] = text.count('?')
# Count of positive emojis
positive_emojis = ['😊', '😃', '😍', '👍', '❤️', '😄', '😁', '😆']
features['positive_emoji_count'] = sum(text.count(emoji) for emoji in positive_emojis)
# Count of negative emojis
negative_emojis = ['😞', '😠', '👎', '😢', '😭', '😤', '😡']
features['negative_emoji_count'] = sum(text.count(emoji) for emoji in negative_emojis)
# Check for all caps words (shouting)
words = text.split()
features['all_caps_count'] = sum(1 for word in words if word.isupper() and len(word) > 1)
# Word count
features['word_count'] = len(words)
# Character count
features['char_count'] = len(text)
# Average word length
features['avg_word_length'] = features['char_count'] / features['word_count'] if features['word_count'] > 0 else 0
return features
def preprocess(self, text, advanced=True):
"""Complete preprocessing pipeline"""
# Convert emojis
text = self.convert_emojis(text)
# Clean text
cleaned = self.clean_text(text)
# Tokenize
tokens = self.tokenize(cleaned)
# Remove stopwords
tokens = self.remove_stopwords(tokens)
# Lemmatize
tokens = self.lemmatize(tokens)
# Extract additional features if advanced preprocessing is enabled
if advanced:
sentiment_features = self.extract_sentiment_features(text)
return ' '.join(tokens), sentiment_features
return ' '.join(tokens)
def preprocess_batch(self, texts, advanced=True):
"""Preprocess a batch of texts"""
results = []
for text in texts:
results.append(self.preprocess(text, advanced))
return results

7. utils/feature_extraction.py

from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer
from sklearn.feature_extraction.text import TfidfTransformer
import numpy as np
from textblob import TextBlob
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
class FeatureExtractor:
def __init__(self):
self.count_vectorizer = CountVectorizer(max_features=5000, ngram_range=(1, 3))
self.tfidf_vectorizer = TfidfVectorizer(max_features=5000, ngram_range=(1, 3))
self.vader_analyzer = SentimentIntensityAnalyzer()
def extract_bow_features(self, texts, fit=False):
"""Extract Bag of Words features"""
if fit:
features = self.count_vectorizer.fit_transform(texts)
self.bow_feature_names = self.count_vectorizer.get_feature_names_out()
else:
features = self.count_vectorizer.transform(texts)
return features
def extract_tfidf_features(self, texts, fit=False):
"""Extract TF-IDF features"""
if fit:
features = self.tfidf_vectorizer.fit_transform(texts)
self.tfidf_feature_names = self.tfidf_vectorizer.get_feature_names_out()
else:
features = self.tfidf_vectorizer.transform(texts)
return features
def extract_lexicon_features(self, texts):
"""Extract lexicon-based sentiment features using TextBlob and VADER"""
features = []
for text in texts:
text_features = {}
# TextBlob features
blob = TextBlob(text)
text_features['textblob_polarity'] = blob.sentiment.polarity
text_features['textblob_subjectivity'] = blob.sentiment.subjectivity
# VADER features
vader_scores = self.vader_analyzer.polarity_scores(text)
text_features['vader_neg'] = vader_scores['neg']
text_features['vader_neu'] = vader_scores['neu']
text_features['vader_pos'] = vader_scores['pos']
text_features['vader_compound'] = vader_scores['compound']
features.append(text_features)
return np.array([list(f.values()) for f in features])
def extract_ngram_features(self, texts, n=3):
"""Extract n-gram features"""
vectorizer = CountVectorizer(ngram_range=(1, n), max_features=1000)
return vectorizer.fit_transform(texts)
def combine_features(self, texts, use_tfidf=True, use_lexicon=True):
"""Combine multiple feature extraction methods"""
features_list = []
# Extract TF-IDF features
if use_tfidf:
tfidf_features = self.extract_tfidf_features(texts, fit=True)
features_list.append(tfidf_features.toarray())
# Extract lexicon features
if use_lexicon:
lexicon_features = self.extract_lexicon_features(texts)
features_list.append(lexicon_features)
# Combine all features
if features_list:
combined_features = np.hstack(features_list)
return combined_features
else:
return None

8. models/sentiment_classifier.py

from sklearn.naive_bayes import MultinomialNB
from sklearn.linear_model import LogisticRegression
from sklearn.svm import SVC
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
from sklearn.metrics import confusion_matrix, classification_report
import numpy as np
import joblib
import time
from utils.text_preprocessing import TextPreprocessor
from utils.feature_extraction import FeatureExtractor
from config.model_config import ModelConfig
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SentimentClassifier:
def __init__(self, model_type='logistic_regression'):
self.model_type = model_type
self.model = None
self.preprocessor = TextPreprocessor()
self.feature_extractor = FeatureExtractor()
self.config = ModelConfig()
self.is_trained = False
# Initialize model based on type
self._initialize_model()
def _initialize_model(self):
"""Initialize the specified model"""
if self.model_type == 'naive_bayes':
self.model = MultinomialNB(alpha=1.0)
elif self.model_type == 'logistic_regression':
self.model = LogisticRegression(
max_iter=1000, 
C=1.0, 
random_state=42,
multi_class='multinomial'
)
elif self.model_type == 'svm':
self.model = SVC(kernel='linear', probability=True, random_state=42)
elif self.model_type == 'random_forest':
self.model = RandomForestClassifier(
n_estimators=100, 
max_depth=10, 
random_state=42
)
elif self.model_type == 'gradient_boosting':
self.model = GradientBoostingClassifier(
n_estimators=100, 
learning_rate=0.1, 
random_state=42
)
else:
raise ValueError(f"Unsupported model type: {self.model_type}")
logger.info(f"Initialized {self.model_type} model")
def train(self, texts, labels, test_size=0.2, random_state=42, use_grid_search=False):
"""Train the sentiment classifier"""
try:
start_time = time.time()
# Preprocess texts
logger.info("Preprocessing texts...")
processed_texts = [self.preprocessor.preprocess(text) for text in texts]
# Extract features
logger.info("Extracting features...")
X = self.feature_extractor.extract_tfidf_features(processed_texts, fit=True)
# Convert labels to numerical values
y = np.array([self.config.LABEL_MAPPING[label.lower()] for label in labels])
# Split data
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=test_size, random_state=random_state, stratify=y
)
# Hyperparameter tuning if requested
if use_grid_search:
self._hyperparameter_tuning(X_train, y_train)
else:
# Train model
logger.info(f"Training {self.model_type} model...")
self.model.fit(X_train, y_train)
# Make predictions
y_pred = self.model.predict(X_test)
y_pred_proba = self.model.predict_proba(X_test) if hasattr(self.model, 'predict_proba') else None
# Calculate metrics
metrics = self._calculate_metrics(y_test, y_pred, y_pred_proba)
# Cross-validation
cv_scores = cross_val_score(self.model, X_train, y_train, cv=5)
metrics['cv_mean'] = cv_scores.mean()
metrics['cv_std'] = cv_scores.std()
# Add training metadata
metrics['model_type'] = self.model_type
metrics['training_time'] = time.time() - start_time
metrics['num_features'] = X.shape[1]
metrics['num_samples'] = len(texts)
self.is_trained = True
logger.info(f"Model trained successfully with accuracy: {metrics['accuracy']:.4f}")
return metrics, X_test, y_test, y_pred
except Exception as e:
logger.error(f"Error during training: {e}")
raise
def _hyperparameter_tuning(self, X_train, y_train):
"""Perform hyperparameter tuning using GridSearchCV"""
logger.info(f"Performing hyperparameter tuning for {self.model_type}...")
param_grids = {
'naive_bayes': {'alpha': [0.1, 0.5, 1.0, 2.0]},
'logistic_regression': {
'C': [0.1, 1.0, 10.0],
'penalty': ['l2']
},
'svm': {
'C': [0.1, 1.0, 10.0],
'kernel': ['linear', 'rbf']
},
'random_forest': {
'n_estimators': [50, 100, 200],
'max_depth': [5, 10, None]
},
'gradient_boosting': {
'n_estimators': [50, 100],
'learning_rate': [0.01, 0.1, 0.3]
}
}
if self.model_type in param_grids:
grid_search = GridSearchCV(
self.model, 
param_grids[self.model_type], 
cv=5, 
scoring='accuracy',
n_jobs=-1
)
grid_search.fit(X_train, y_train)
self.model = grid_search.best_estimator_
logger.info(f"Best parameters: {grid_search.best_params_}")
def _calculate_metrics(self, y_true, y_pred, y_pred_proba=None):
"""Calculate performance metrics"""
metrics = {
'accuracy': accuracy_score(y_true, y_pred),
'precision': precision_score(y_true, y_pred, average='weighted'),
'recall': recall_score(y_true, y_pred, average='weighted'),
'f1_score': f1_score(y_true, y_pred, average='weighted'),
'confusion_matrix': confusion_matrix(y_true, y_pred).tolist(),
'classification_report': classification_report(
y_true, y_pred, 
target_names=list(self.config.SENTIMENT_LABELS.values()),
output_dict=True
)
}
# Add per-class metrics
for i, label in self.config.SENTIMENT_LABELS.items():
metrics[f'{label.lower()}_precision'] = precision_score(
y_true, y_pred, labels=[i], average='macro'
)
metrics[f'{label.lower()}_recall'] = recall_score(
y_true, y_pred, labels=[i], average='macro'
)
return metrics
def predict(self, text):
"""Predict sentiment for a single text"""
if not self.is_trained:
raise ValueError("Model not trained yet. Please train the model first.")
# Preprocess text
processed_text = self.preprocessor.preprocess(text)
# Extract features
X = self.feature_extractor.extract_tfidf_features([processed_text])
# Make prediction
prediction_idx = self.model.predict(X)[0]
prediction = self.config.SENTIMENT_LABELS[prediction_idx]
# Get probability scores if available
confidence = None
probabilities = None
if hasattr(self.model, 'predict_proba'):
probabilities = self.model.predict_proba(X)[0]
confidence = float(max(probabilities))
return {
'sentiment': prediction,
'sentiment_idx': int(prediction_idx),
'confidence': confidence,
'probabilities': probabilities.tolist() if probabilities is not None else None
}
def predict_batch(self, texts):
"""Predict sentiment for multiple texts"""
if not self.is_trained:
raise ValueError("Model not trained yet. Please train the model first.")
# Preprocess texts
processed_texts = [self.preprocessor.preprocess(text) for text in texts]
# Extract features
X = self.feature_extractor.extract_tfidf_features(processed_texts)
# Make predictions
predictions_idx = self.model.predict(X)
predictions = [self.config.SENTIMENT_LABELS[idx] for idx in predictions_idx]
# Get probability scores if available
confidences = None
if hasattr(self.model, 'predict_proba'):
probabilities = self.model.predict_proba(X)
confidences = [float(max(probs)) for probs in probabilities]
results = []
for i, (text, pred_idx, pred) in enumerate(zip(texts, predictions_idx, predictions)):
result = {
'text': text,
'sentiment': pred,
'sentiment_idx': int(pred_idx),
'confidence': confidences[i] if confidences else None
}
results.append(result)
return results
def save_model(self, filepath='models/saved_models/sentiment_model.pkl'):
"""Save the trained model to disk"""
if not self.is_trained:
raise ValueError("No trained model to save.")
model_data = {
'model': self.model,
'model_type': self.model_type,
'feature_extractor': self.feature_extractor,
'preprocessor': self.preprocessor,
'config': self.config
}
joblib.dump(model_data, filepath)
logger.info(f"Model saved to {filepath}")
def load_model(self, filepath='models/saved_models/sentiment_model.pkl'):
"""Load a trained model from disk"""
try:
model_data = joblib.load(filepath)
self.model = model_data['model']
self.model_type = model_data['model_type']
self.feature_extractor = model_data['feature_extractor']
self.preprocessor = model_data['preprocessor']
self.config = model_data['config']
self.is_trained = True
logger.info(f"Model loaded from {filepath}")
except Exception as e:
logger.error(f"Error loading model: {e}")
raise
def get_model_info(self):
"""Get information about the current model"""
return {
'is_trained': self.is_trained,
'model_type': self.model_type,
'num_features': len(self.feature_extractor.tfidf_feature_names) if hasattr(self.feature_extractor, 'tfidf_feature_names') else 0,
'sentiment_labels': self.config.SENTIMENT_LABELS
}

9. models/deep_learning_model.py

import tensorflow as tf
from tensorflow.keras.models import Sequential, Model
from tensorflow.keras.layers import Embedding, LSTM, Dense, Dropout, Bidirectional
from tensorflow.keras.layers import GlobalMaxPooling1D, Conv1D, MaxPooling1D
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint
from sklearn.model_selection import train_test_split
import numpy as np
import joblib
from config.model_config import ModelConfig
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DeepLearningSentimentClassifier:
def __init__(self, model_type='lstm', max_words=5000, max_len=200):
self.model_type = model_type
self.max_words = max_words
self.max_len = max_len
self.model = None
self.tokenizer = Tokenizer(num_words=max_words)
self.config = ModelConfig()
self.is_trained = False
def build_model(self):
"""Build deep learning model architecture"""
if self.model_type == 'lstm':
self.model = Sequential([
Embedding(self.max_words, 128, input_length=self.max_len),
LSTM(128, dropout=0.2, recurrent_dropout=0.2, return_sequences=True),
LSTM(64, dropout=0.2, recurrent_dropout=0.2),
Dense(64, activation='relu'),
Dropout(0.5),
Dense(3, activation='softmax')
])
elif self.model_type == 'bilstm':
self.model = Sequential([
Embedding(self.max_words, 128, input_length=self.max_len),
Bidirectional(LSTM(128, dropout=0.2, recurrent_dropout=0.2)),
Dense(64, activation='relu'),
Dropout(0.5),
Dense(3, activation='softmax')
])
elif self.model_type == 'cnn_lstm':
self.model = Sequential([
Embedding(self.max_words, 128, input_length=self.max_len),
Conv1D(128, 5, activation='relu'),
MaxPooling1D(5),
LSTM(64, dropout=0.2, recurrent_dropout=0.2),
Dense(64, activation='relu'),
Dropout(0.5),
Dense(3, activation='softmax')
])
# Compile model
self.model.compile(
optimizer='adam',
loss='sparse_categorical_crossentropy',
metrics=['accuracy']
)
logger.info(f"Built {self.model_type} model architecture")
logger.info(self.model.summary())
def prepare_data(self, texts, labels):
"""Prepare data for deep learning model"""
# Tokenize texts
self.tokenizer.fit_on_texts(texts)
sequences = self.tokenizer.texts_to_sequences(texts)
# Pad sequences
X = pad_sequences(sequences, maxlen=self.max_len, padding='post', truncating='post')
# Convert labels
y = np.array([self.config.LABEL_MAPPING[label.lower()] for label in labels])
return X, y
def train(self, texts, labels, validation_split=0.2, epochs=10, batch_size=32):
"""Train the deep learning model"""
try:
# Prepare data
X, y = self.prepare_data(texts, labels)
# Split data
X_train, X_val, y_train, y_val = train_test_split(
X, y, test_size=validation_split, random_state=42, stratify=y
)
# Build model if not already built
if self.model is None:
self.build_model()
# Callbacks
callbacks = [
EarlyStopping(patience=3, restore_best_weights=True),
ModelCheckpoint(
'models/saved_models/best_dl_model.h5',
save_best_only=True,
monitor='val_accuracy'
)
]
# Train model
logger.info(f"Training {self.model_type} model...")
history = self.model.fit(
X_train, y_train,
validation_data=(X_val, y_val),
epochs=epochs,
batch_size=batch_size,
callbacks=callbacks,
verbose=1
)
self.is_trained = True
# Evaluate on validation set
val_loss, val_accuracy = self.model.evaluate(X_val, y_val, verbose=0)
metrics = {
'model_type': self.model_type,
'val_accuracy': val_accuracy,
'val_loss': val_loss,
'history': {
'accuracy': history.history['accuracy'],
'val_accuracy': history.history['val_accuracy'],
'loss': history.history['loss'],
'val_loss': history.history['val_loss']
}
}
logger.info(f"Model trained successfully with validation accuracy: {val_accuracy:.4f}")
return metrics, X_val, y_val
except Exception as e:
logger.error(f"Error during training: {e}")
raise
def predict(self, text):
"""Predict sentiment for a single text"""
if not self.is_trained:
raise ValueError("Model not trained yet. Please train the model first.")
# Prepare text
sequences = self.tokenizer.texts_to_sequences([text])
X = pad_sequences(sequences, maxlen=self.max_len, padding='post', truncating='post')
# Make prediction
probabilities = self.model.predict(X)[0]
prediction_idx = np.argmax(probabilities)
prediction = self.config.SENTIMENT_LABELS[prediction_idx]
confidence = float(probabilities[prediction_idx])
return {
'sentiment': prediction,
'sentiment_idx': int(prediction_idx),
'confidence': confidence,
'probabilities': probabilities.tolist()
}
def predict_batch(self, texts):
"""Predict sentiment for multiple texts"""
if not self.is_trained:
raise ValueError("Model not trained yet. Please train the model first.")
# Prepare texts
sequences = self.tokenizer.texts_to_sequences(texts)
X = pad_sequences(sequences, maxlen=self.max_len, padding='post', truncating='post')
# Make predictions
probabilities = self.model.predict(X)
predictions_idx = np.argmax(probabilities, axis=1)
predictions = [self.config.SENTIMENT_LABELS[idx] for idx in predictions_idx]
confidences = [float(probs[idx]) for probs, idx in zip(probabilities, predictions_idx)]
results = []
for i, (text, pred_idx, pred, conf) in enumerate(zip(texts, predictions_idx, predictions, confidences)):
results.append({
'text': text,
'sentiment': pred,
'sentiment_idx': int(pred_idx),
'confidence': conf,
'probabilities': probabilities[i].tolist()
})
return results
def save_model(self, filepath='models/saved_models/dl_sentiment_model'):
"""Save the trained model and tokenizer"""
if not self.is_trained:
raise ValueError("No trained model to save.")
# Save Keras model
self.model.save(f"{filepath}.h5")
# Save tokenizer and config
model_data = {
'tokenizer': self.tokenizer,
'model_type': self.model_type,
'max_words': self.max_words,
'max_len': self.max_len,
'config': self.config
}
joblib.dump(model_data, f"{filepath}_config.pkl")
logger.info(f"Model saved to {filepath}")
def load_model(self, filepath='models/saved_models/dl_sentiment_model'):
"""Load a trained model from disk"""
try:
# Load Keras model
self.model = tf.keras.models.load_model(f"{filepath}.h5")
# Load tokenizer and config
model_data = joblib.load(f"{filepath}_config.pkl")
self.tokenizer = model_data['tokenizer']
self.model_type = model_data['model_type']
self.max_words = model_data['max_words']
self.max_len = model_data['max_len']
self.config = model_data['config']
self.is_trained = True
logger.info(f"Model loaded from {filepath}")
except Exception as e:
logger.error(f"Error loading model: {e}")
raise

10. models/model_comparator.py

from models.sentiment_classifier import SentimentClassifier
from models.deep_learning_model import DeepLearningSentimentClassifier
from database.db_operations import MongoDBOperations
from utils.visualization import Visualizer
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ModelComparator:
def __init__(self):
self.models = {}
self.results = {}
self.db_ops = MongoDBOperations()
self.visualizer = Visualizer()
def initialize_models(self):
"""Initialize all available models"""
# Traditional ML models
self.models['naive_bayes'] = SentimentClassifier('naive_bayes')
self.models['logistic_regression'] = SentimentClassifier('logistic_regression')
self.models['svm'] = SentimentClassifier('svm')
self.models['random_forest'] = SentimentClassifier('random_forest')
# Deep Learning models
self.models['lstm'] = DeepLearningSentimentClassifier('lstm')
self.models['bilstm'] = DeepLearningSentimentClassifier('bilstm')
logger.info(f"Initialized {len(self.models)} models")
def train_and_compare(self, texts, labels, test_size=0.2):
"""Train all models and compare their performance"""
results = {}
# Split data for consistent evaluation
X_train, X_test, y_train, y_test = train_test_split(
texts, labels, test_size=test_size, random_state=42, stratify=labels
)
for model_name, model in self.models.items():
logger.info(f"\n{'='*50}")
logger.info(f"Training {model_name}...")
logger.info(f"{'='*50}")
start_time = time.time()
try:
# Train model
if 'lstm' in model_name:  # Deep learning models
metrics, _, _ = model.train(X_train, y_train, epochs=5)
# Test on test set
predictions = model.predict_batch(X_test)
y_pred = [p['sentiment_idx'] for p in predictions]
else:  # Traditional ML models
metrics, X_test_tfidf, y_test, y_pred = model.train(
X_train, y_train, test_size=test_size
)
training_time = time.time() - start_time
# Store results
results[model_name] = {
'model': model,
'metrics': metrics,
'training_time': training_time,
'predictions': y_pred if 'y_pred' in locals() else None
}
logger.info(f"{model_name} completed in {training_time:.2f} seconds")
except Exception as e:
logger.error(f"Error training {model_name}: {e}")
results[model_name] = {'error': str(e)}
self.results = results
return results
def get_comparison_report(self):
"""Generate comparison report for all models"""
comparison = []
for model_name, result in self.results.items():
if 'error' in result:
continue
metrics = result['metrics']
if isinstance(metrics, dict):
comparison.append({
'Model': model_name,
'Accuracy': metrics.get('accuracy', metrics.get('val_accuracy', 0)),
'Precision': metrics.get('precision', 0),
'Recall': metrics.get('recall', 0),
'F1-Score': metrics.get('f1_score', 0),
'Training Time (s)': round(result['training_time'], 2)
})
df = pd.DataFrame(comparison)
df = df.sort_values('Accuracy', ascending=False)
return df
def plot_comparison(self):
"""Plot comparison of model performances"""
df = self.get_comparison_report()
if not df.empty:
# Plot accuracy comparison
self.visualizer.plot_model_comparison(
df['Model'].tolist(),
df['Accuracy'].tolist(),
'Model Accuracy Comparison',
'Accuracy'
)
# Plot training time comparison
self.visualizer.plot_model_comparison(
df['Model'].tolist(),
df['Training Time (s)'].tolist(),
'Model Training Time Comparison',
'Training Time (seconds)'
)
def get_best_model(self, metric='Accuracy'):
"""Get the best performing model based on specified metric"""
df = self.get_comparison_report()
if not df.empty:
best_model = df.loc[df[metric].idxmax()]
return {
'model_name': best_model['Model'],
'metric_value': best_model[metric],
'metric': metric
}
return None
def save_all_models(self):
"""Save all trained models"""
for model_name, result in self.results.items():
if 'error' not in result and 'model' in result:
try:
result['model'].save_model(f'models/saved_models/{model_name}_model')
logger.info(f"Saved {model_name} model")
except Exception as e:
logger.error(f"Error saving {model_name}: {e}")

11. utils/visualization.py

import matplotlib.pyplot as plt
import seaborn as sns
import plotly.graph_objects as go
import plotly.express as px
from wordcloud import WordCloud
import numpy as np
import pandas as pd
from sklearn.metrics import confusion_matrix
import io
import base64
class Visualizer:
def __init__(self):
plt.style.use('seaborn-v0_8-darkgrid')
sns.set_palette("husl")
def plot_sentiment_distribution(self, sentiments, title="Sentiment Distribution"):
"""Plot distribution of sentiments"""
plt.figure(figsize=(10, 6))
# Count sentiments
sentiment_counts = pd.Series(sentiments).value_counts()
# Create bar plot
ax = sns.barplot(x=sentiment_counts.index, y=sentiment_counts.values)
plt.title(title, fontsize=16, fontweight='bold')
plt.xlabel('Sentiment', fontsize=12)
plt.ylabel('Count', fontsize=12)
# Add value labels on bars
for i, v in enumerate(sentiment_counts.values):
ax.text(i, v + 0.1, str(v), ha='center', fontsize=11)
plt.tight_layout()
# Convert to base64 for web display
img = io.BytesIO()
plt.savefig(img, format='png', dpi=100, bbox_inches='tight')
img.seek(0)
plot_url = base64.b64encode(img.getvalue()).decode()
plt.close()
return plot_url
def plot_confusion_matrix(self, y_true, y_pred, labels=['Negative', 'Neutral', 'Positive']):
"""Plot confusion matrix"""
cm = confusion_matrix(y_true, y_pred)
plt.figure(figsize=(8, 6))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
xticklabels=labels, yticklabels=labels)
plt.title('Confusion Matrix', fontsize=16, fontweight='bold')
plt.ylabel('Actual', fontsize=12)
plt.xlabel('Predicted', fontsize=12)
plt.tight_layout()
# Convert to base64
img = io.BytesIO()
plt.savefig(img, format='png', dpi=100, bbox_inches='tight')
img.seek(0)
plot_url = base64.b64encode(img.getvalue()).decode()
plt.close()
return plot_url
def create_wordcloud(self, texts, title="Word Cloud"):
"""Create word cloud from texts"""
# Combine all texts
all_text = ' '.join(texts)
# Create word cloud
wordcloud = WordCloud(
width=800, height=400,
background_color='white',
max_words=100,
colormap='viridis'
).generate(all_text)
plt.figure(figsize=(12, 6))
plt.imshow(wordcloud, interpolation='bilinear')
plt.title(title, fontsize=16, fontweight='bold')
plt.axis('off')
plt.tight_layout()
# Convert to base64
img = io.BytesIO()
plt.savefig(img, format='png', dpi=100, bbox_inches='tight')
img.seek(0)
plot_url = base64.b64encode(img.getvalue()).decode()
plt.close()
return plot_url
def plot_interactive_sentiment_timeline(self, dates, sentiments, title="Sentiment Timeline"):
"""Create interactive timeline plot using plotly"""
df = pd.DataFrame({
'Date': dates,
'Sentiment': sentiments
})
fig = px.line(df, x='Date', y='Sentiment', title=title)
fig.update_layout(
xaxis_title="Date",
yaxis_title="Sentiment Score",
hovermode='x'
)
return fig.to_html()
def plot_model_comparison(self, model_names, scores, title, ylabel):
"""Plot comparison of different models"""
plt.figure(figsize=(12, 6))
colors = plt.cm.viridis(np.linspace(0, 1, len(model_names)))
bars = plt.bar(model_names, scores, color=colors)
plt.title(title, fontsize=16, fontweight='bold')
plt.xlabel('Model', fontsize=12)
plt.ylabel(ylabel, fontsize=12)
plt.xticks(rotation=45, ha='right')
# Add value labels
for bar, score in zip(bars, scores):
plt.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.01,
f'{score:.3f}', ha='center', va='bottom', fontsize=10)
plt.tight_layout()
# Convert to base64
img = io.BytesIO()
plt.savefig(img, format='png', dpi=100, bbox_inches='tight')
img.seek(0)
plot_url = base64.b64encode(img.getvalue()).decode()
plt.close()
return plot_url
def plot_feature_importance(self, feature_names, importance_scores, top_n=20):
"""Plot feature importance"""
# Sort by importance
indices = np.argsort(importance_scores)[-top_n:]
plt.figure(figsize=(10, 8))
plt.barh(range(top_n), importance_scores[indices])
plt.yticks(range(top_n), [feature_names[i] for i in indices])
plt.xlabel('Importance Score', fontsize=12)
plt.title(f'Top {top_n} Most Important Features', fontsize=16, fontweight='bold')
plt.tight_layout()
# Convert to base64
img = io.BytesIO()
plt.savefig(img, format='png', dpi=100, bbox_inches='tight')
img.seek(0)
plot_url = base64.b64encode(img.getvalue()).decode()
plt.close()
return plot_url
def plot_sentiment_confidence(self, predictions, confidences):
"""Plot sentiment predictions with confidence scores"""
plt.figure(figsize=(12, 6))
# Create scatter plot
colors = {'Positive': 'green', 'Neutral': 'blue', 'Negative': 'red'}
point_colors = [colors[p] for p in predictions]
plt.scatter(range(len(predictions)), confidences, 
c=point_colors, alpha=0.6, s=100)
plt.xlabel('Sample Index', fontsize=12)
plt.ylabel('Confidence Score', fontsize=12)
plt.title('Prediction Confidence by Sentiment', fontsize=16, fontweight='bold')
plt.ylim(0, 1.1)
# Add legend
from matplotlib.patches import Patch
legend_elements = [Patch(facecolor=color, label=sentiment) 
for sentiment, color in colors.items()]
plt.legend(handles=legend_elements)
plt.tight_layout()
# Convert to base64
img = io.BytesIO()
plt.savefig(img, format='png', dpi=100, bbox_inches='tight')
img.seek(0)
plot_url = base64.b64encode(img.getvalue()).decode()
plt.close()
return plot_url

12. api/app.py (Flask API)

from flask import Flask, request, jsonify, render_template
from flask_cors import CORS
from flask_restful import Api, Resource
from models.sentiment_classifier import SentimentClassifier
from models.deep_learning_model import DeepLearningSentimentClassifier
from models.model_comparator import ModelComparator
from database.db_operations import MongoDBOperations
from utils.text_preprocessing import TextPreprocessor
from utils.visualization import Visualizer
import pandas as pd
import json
from bson import ObjectId
import logging
app = Flask(__name__)
CORS(app)
api = Api(app)
# Initialize components
db_ops = MongoDBOperations()
preprocessor = TextPreprocessor()
visualizer = Visualizer()
classifier = SentimentClassifier('logistic_regression')
dl_classifier = DeepLearningSentimentClassifier('lstm')
comparator = ModelComparator()
# Try to load pre-trained models
try:
classifier.load_model()
dl_classifier.load_model()
logging.info("Loaded pre-trained models")
except:
logging.info("No pre-trained models found. Please train models first.")
class JSONEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, ObjectId):
return str(o)
return json.JSONEncoder.default(self, o)
app.json_encoder = JSONEncoder
class HealthCheck(Resource):
def get(self):
return {'status': 'healthy', 'message': 'Sentiment Analyzer API is running'}
class AnalyzeSentiment(Resource):
def post(self):
data = request.get_json()
text = data.get('text', '')
model_type = data.get('model', 'logistic_regression')
if not text:
return {'error': 'No text provided'}, 400
try:
# Choose model
if model_type == 'lstm' or model_type == 'bilstm':
result = dl_classifier.predict(text)
else:
result = classifier.predict(text)
# Save to database
review_id = db_ops.insert_review(text)
db_ops.save_prediction(
review_id, text, result['sentiment'], 
result['confidence'], model_used=model_type
)
return {
'success': True,
'result': result
}
except Exception as e:
return {'error': str(e)}, 500
class BatchAnalyze(Resource):
def post(self):
data = request.get_json()
texts = data.get('texts', [])
model_type = data.get('model', 'logistic_regression')
if not texts:
return {'error': 'No texts provided'}, 400
try:
# Choose model
if model_type == 'lstm' or model_type == 'bilstm':
results = dl_classifier.predict_batch(texts)
else:
results = classifier.predict_batch(texts)
# Save to database
for text, result in zip(texts, results):
review_id = db_ops.insert_review(text)
db_ops.save_prediction(
review_id, text, result['sentiment'],
result['confidence'], model_used=model_type
)
return {
'success': True,
'results': results,
'count': len(results)
}
except Exception as e:
return {'error': str(e)}, 500
class TrainModel(Resource):
def post(self):
data = request.get_json()
model_type = data.get('model_type', 'logistic_regression')
try:
# Get training data
texts, labels = db_ops.get_training_data()
if len(texts) == 0:
return {'error': 'No training data found'}, 400
# Train appropriate model
if model_type == 'lstm' or model_type == 'bilstm':
model = DeepLearningSentimentClassifier(model_type)
metrics, _, _ = model.train(texts, labels)
model.save_model()
else:
model = SentimentClassifier(model_type)
metrics, _, _, _ = model.train(texts, labels)
model.save_model()
# Save metrics to database
db_ops.update_model_metrics(metrics)
return {
'success': True,
'message': f'{model_type} model trained successfully',
'metrics': metrics
}
except Exception as e:
return {'error': str(e)}, 500
class CompareModels(Resource):
def get(self):
try:
# Get training data
texts, labels = db_ops.get_training_data()
if len(texts) == 0:
return {'error': 'No training data found'}, 400
# Initialize comparator
comparator.initialize_models()
# Train and compare
results = comparator.train_and_compare(texts, labels)
# Get comparison report
comparison_df = comparator.get_comparison_report()
return {
'success': True,
'comparison': comparison_df.to_dict('records'),
'best_model': comparator.get_best_model()
}
except Exception as e:
return {'error': str(e)}, 500
class GetStatistics(Resource):
def get(self):
try:
# Get database statistics
prediction_stats = db_ops.get_prediction_statistics()
model_performance = db_ops.get_model_performance()
# Get reviews data
df = db_ops.export_reviews_to_dataframe()
statistics = {
'total_reviews': len(df) if not df.empty else 0,
'sentiment_distribution': df['sentiment_label'].value_counts().to_dict() if not df.empty and 'sentiment_label' in df.columns else {},
'prediction_statistics': prediction_stats,
'model_performance': model_performance
}
return {
'success': True,
'statistics': statistics
}
except Exception as e:
return {'error': str(e)}, 500
class GetVisualizations(Resource):
def get(self):
try:
# Get data for visualizations
df = db_ops.export_reviews_to_dataframe()
if df.empty or 'sentiment_label' not in df.columns:
return {'error': 'No data available for visualization'}, 400
# Generate visualizations
plots = {}
# Sentiment distribution
plots['sentiment_distribution'] = visualizer.plot_sentiment_distribution(
df['sentiment_label'].tolist()
)
# Word cloud
plots['wordcloud'] = visualizer.create_wordcloud(
df['review_text'].tolist()
)
return {
'success': True,
'plots': plots
}
except Exception as e:
return {'error': str(e)}, 500
# Web routes
@app.route('/')
def index():
return render_template('index.html')
@app.route('/analyze')
def analyze_page():
return render_template('analyze.html')
@app.route('/dashboard')
def dashboard():
return render_template('dashboard.html')
# API routes
api.add_resource(HealthCheck, '/api/health')
api.add_resource(AnalyzeSentiment, '/api/analyze')
api.add_resource(BatchAnalyze, '/api/analyze/batch')
api.add_resource(TrainModel, '/api/train')
api.add_resource(CompareModels, '/api/compare')
api.add_resource(GetStatistics, '/api/statistics')
api.add_resource(GetVisualizations, '/api/visualizations')
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=5000)

13. web/templates/index.html

<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Sentiment Analyzer - Home</title>
<link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
<link href="https://cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css" rel="stylesheet">
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.0.0/css/all.min.css">
</head>
<body>
<nav class="navbar navbar-expand-lg navbar-dark bg-primary">
<div class="container">
<a class="navbar-brand" href="/">
<i class="fas fa-smile"></i> Sentiment Analyzer
</a>
<button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target="#navbarNav">
<span class="navbar-toggler-icon"></span>
</button>
<div class="collapse navbar-collapse" id="navbarNav">
<ul class="navbar-nav ms-auto">
<li class="nav-item">
<a class="nav-link active" href="/">Home</a>
</li>
<li class="nav-item">
<a class="nav-link" href="/analyze">Analyze</a>
</li>
<li class="nav-item">
<a class="nav-link" href="/dashboard">Dashboard</a>
</li>
</ul>
</div>
</div>
</nav>
<div class="hero-section">
<div class="container text-center">
<h1 class="display-4">Sentiment Analysis Made Easy</h1>
<p class="lead">Analyze movie and product reviews with state-of-the-art machine learning</p>
<a href="/analyze" class="btn btn-light btn-lg mt-3">
<i class="fas fa-play"></i> Start Analyzing
</a>
</div>
</div>
<div class="container features-section py-5">
<h2 class="text-center mb-5">Key Features</h2>
<div class="row">
<div class="col-md-4 mb-4">
<div class="card h-100">
<div class="card-body text-center">
<i class="fas fa-robot fa-3x text-primary mb-3"></i>
<h5 class="card-title">Multiple ML Models</h5>
<p class="card-text">Choose from Naive Bayes, Logistic Regression, SVM, LSTM, and more</p>
</div>
</div>
</div>
<div class="col-md-4 mb-4">
<div class="card h-100">
<div class="card-body text-center">
<i class="fas fa-chart-line fa-3x text-primary mb-3"></i>
<h5 class="card-title">Real-time Analysis</h5>
<p class="card-text">Get instant sentiment predictions with confidence scores</p>
</div>
</div>
</div>
<div class="col-md-4 mb-4">
<div class="card h-100">
<div class="card-body text-center">
<i class="fas fa-database fa-3x text-primary mb-3"></i>
<h5 class="card-title">MongoDB Integration</h5>
<p class="card-text">All reviews and predictions are stored for future analysis</p>
</div>
</div>
</div>
</div>
</div>
<div class="bg-light py-5">
<div class="container">
<h2 class="text-center mb-5">How It Works</h2>
<div class="row">
<div class="col-md-3 text-center">
<div class="step-circle">1</div>
<h5>Input Text</h5>
<p>Enter your movie or product review</p>
</div>
<div class="col-md-3 text-center">
<div class="step-circle">2</div>
<h5>Preprocessing</h5>
<p>Text is cleaned and prepared for analysis</p>
</div>
<div class="col-md-3 text-center">
<div class="step-circle">3</div>
<h5>ML Analysis</h5>
<p>Our models analyze the sentiment</p>
</div>
<div class="col-md-3 text-center">
<div class="step-circle">4</div>
<h5>Get Results</h5>
<p>Receive sentiment classification with confidence score</p>
</div>
</div>
</div>
</div>
<footer class="bg-dark text-white py-4">
<div class="container text-center">
<p>&copy; 2024 Sentiment Analyzer. All rights reserved.</p>
</div>
</footer>
<script src="https://cdn.jsdelivr.net/npm/[email protected]/dist/js/bootstrap.bundle.min.js"></script>
<script src="{{ url_for('static', filename='js/main.js') }}"></script>
</body>
</html>

14. web/templates/analyze.html

<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Sentiment Analyzer - Analyze</title>
<link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
<link href="https://cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css" rel="stylesheet">
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.0.0/css/all.min.css">
</head>
<body>
<nav class="navbar navbar-expand-lg navbar-dark bg-primary">
<div class="container">
<a class="navbar-brand" href="/">
<i class="fas fa-smile"></i> Sentiment Analyzer
</a>
<button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target="#navbarNav">
<span class="navbar-toggler-icon"></span>
</button>
<div class="collapse navbar-collapse" id="navbarNav">
<ul class="navbar-nav ms-auto">
<li class="nav-item">
<a class="nav-link" href="/">Home</a>
</li>
<li class="nav-item">
<a class="nav-link active" href="/analyze">Analyze</a>
</li>
<li class="nav-item">
<a class="nav-link" href="/dashboard">Dashboard</a>
</li>
</ul>
</div>
</div>
</nav>
<div class="container py-5">
<h2 class="text-center mb-4">Analyze Sentiment</h2>
<div class="row">
<div class="col-md-8 mx-auto">
<div class="card">
<div class="card-body">
<form id="analyzeForm">
<div class="mb-3">
<label for="reviewText" class="form-label">Enter your review:</label>
<textarea class="form-control" id="reviewText" rows="4" 
placeholder="Type or paste your movie or product review here..."></textarea>
</div>
<div class="mb-3">
<label for="modelSelect" class="form-label">Select Model:</label>
<select class="form-select" id="modelSelect">
<option value="logistic_regression">Logistic Regression</option>
<option value="naive_bayes">Naive Bayes</option>
<option value="svm">SVM</option>
<option value="random_forest">Random Forest</option>
<option value="lstm">LSTM (Deep Learning)</option>
<option value="bilstm">Bi-LSTM (Deep Learning)</option>
</select>
</div>
<button type="submit" class="btn btn-primary w-100">
<i class="fas fa-search"></i> Analyze Sentiment
</button>
</form>
</div>
</div>
<!-- Results Section -->
<div id="resultsSection" class="mt-4" style="display: none;">
<div class="card">
<div class="card-header bg-success text-white">
<h5 class="mb-0">Analysis Results</h5>
</div>
<div class="card-body">
<div class="row">
<div class="col-md-6">
<h6>Sentiment:</h6>
<div id="sentimentResult" class="display-4"></div>
</div>
<div class="col-md-6">
<h6>Confidence:</h6>
<div id="confidenceResult" class="display-4"></div>
</div>
</div>
<div class="mt-4">
<h6>Probability Distribution:</h6>
<div class="progress mb-2" id="negativeBar">
<div class="progress-bar bg-danger" role="progressbar" style="width: 0%">Negative</div>
</div>
<div class="progress mb-2" id="neutralBar">
<div class="progress-bar bg-warning" role="progressbar" style="width: 0%">Neutral</div>
</div>
<div class="progress mb-2" id="positiveBar">
<div class="progress-bar bg-success" role="progressbar" style="width: 0%">Positive</div>
</div>
</div>
<div class="mt-4">
<h6>Processed Text:</h6>
<p id="processedText" class="text-muted"></p>
</div>
</div>
</div>
</div>
<!-- Error Section -->
<div id="errorSection" class="mt-4" style="display: none;">
<div class="alert alert-danger" role="alert">
<i class="fas fa-exclamation-triangle"></i> 
<span id="errorMessage"></span>
</div>
</div>
<!-- Loading Spinner -->
<div id="loadingSpinner" class="text-center mt-4" style="display: none;">
<div class="spinner-border text-primary" role="status">
<span class="visually-hidden">Loading...</span>
</div>
</div>
</div>
</div>
</div>
<footer class="bg-dark text-white py-4 mt-5">
<div class="container text-center">
<p>&copy; 2024 Sentiment Analyzer. All rights reserved.</p>
</div>
</footer>
<script src="https://cdn.jsdelivr.net/npm/[email protected]/dist/js/bootstrap.bundle.min.js"></script>
<script src="{{ url_for('static', filename='js/main.js') }}"></script>
<script>
document.getElementById('analyzeForm').addEventListener('submit', async (e) => {
e.preventDefault();
const text = document.getElementById('reviewText').value;
const model = document.getElementById('modelSelect').value;
if (!text) {
alert('Please enter a review');
return;
}
// Show loading spinner
document.getElementById('loadingSpinner').style.display = 'block';
document.getElementById('resultsSection').style.display = 'none';
document.getElementById('errorSection').style.display = 'none';
try {
const response = await fetch('/api/analyze', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({ text, model })
});
const data = await response.json();
// Hide loading spinner
document.getElementById('loadingSpinner').style.display = 'none';
if (data.success) {
// Display results
const result = data.result;
document.getElementById('sentimentResult').textContent = result.sentiment;
document.getElementById('confidenceResult').textContent = 
(result.confidence * 100).toFixed(2) + '%';
// Update progress bars if probabilities available
if (result.probabilities) {
const probs = result.probabilities;
document.querySelector('#negativeBar .progress-bar').style.width = 
(probs[0] * 100) + '%';
document.querySelector('#neutralBar .progress-bar').style.width = 
(probs[1] * 100) + '%';
document.querySelector('#positiveBar .progress-bar').style.width = 
(probs[2] * 100) + '%';
}
document.getElementById('resultsSection').style.display = 'block';
} else {
document.getElementById('errorMessage').textContent = data.error;
document.getElementById('errorSection').style.display = 'block';
}
} catch (error) {
document.getElementById('loadingSpinner').style.display = 'none';
document.getElementById('errorMessage').textContent = error.message;
document.getElementById('errorSection').style.display = 'block';
}
});
</script>
</body>
</html>

15. web/static/css/style.css

:root {
--primary-color: #4a90e2;
--secondary-color: #f5f5f5;
--success-color: #28a745;
--danger-color: #dc3545;
--warning-color: #ffc107;
}
body {
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
background-color: #f8f9fa;
}
.hero-section {
background: linear-gradient(135deg, var(--primary-color), #6c5ce7);
color: white;
padding: 100px 0;
}
.features-section {
background-color: white;
}
.features-section .card {
transition: transform 0.3s ease;
border: none;
box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
}
.features-section .card:hover {
transform: translateY(-10px);
}
.step-circle {
width: 50px;
height: 50px;
background-color: var(--primary-color);
color: white;
border-radius: 50%;
display: flex;
align-items: center;
justify-content: center;
margin: 0 auto 20px;
font-size: 24px;
font-weight: bold;
}
/* Dashboard Styles */
.stat-card {
background: linear-gradient(135deg, #667eea, #764ba2);
color: white;
border-radius: 10px;
padding: 20px;
margin-bottom: 20px;
box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
}
.stat-card i {
font-size: 48px;
opacity: 0.8;
}
.stat-card .stat-value {
font-size: 32px;
font-weight: bold;
margin-top: 10px;
}
.stat-card .stat-label {
font-size: 16px;
opacity: 0.9;
}
/* Chart Containers */
.chart-container {
background-color: white;
border-radius: 10px;
padding: 20px;
margin-bottom: 20px;
box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
}
/* Sentiment Badges */
.sentiment-badge {
padding: 5px 10px;
border-radius: 20px;
font-size: 14px;
font-weight: 500;
}
.sentiment-positive {
background-color: #d4edda;
color: #155724;
}
.sentiment-neutral {
background-color: #fff3cd;
color: #856404;
}
.sentiment-negative {
background-color: #f8d7da;
color: #721c24;
}
/* Progress Bars */
.progress {
height: 25px;
margin-bottom: 10px;
border-radius: 5px;
}
.progress-bar {
line-height: 25px;
font-size: 14px;
}
/* Loading Spinner */
.spinner-border {
width: 3rem;
height: 3rem;
}
/* Responsive Design */
@media (max-width: 768px) {
.hero-section {
padding: 50px 0;
}
.hero-section h1 {
font-size: 28px;
}
.stat-card i {
font-size: 36px;
}
.stat-card .stat-value {
font-size: 24px;
}
}
/* Animations */
@keyframes fadeIn {
from {
opacity: 0;
transform: translateY(20px);
}
to {
opacity: 1;
transform: translateY(0);
}
}
.fade-in {
animation: fadeIn 0.5s ease;
}
/* Custom Scrollbar */
::-webkit-scrollbar {
width: 10px;
}
::-webkit-scrollbar-track {
background: #f1f1f1;
}
::-webkit-scrollbar-thumb {
background: var(--primary-color);
border-radius: 5px;
}
::-webkit-scrollbar-thumb:hover {
background: #357abd;
}

16. web/static/js/main.js

// Utility functions
class SentimentAnalyzer {
static async analyzeText(text, model = 'logistic_regression') {
try {
const response = await fetch('/api/analyze', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({ text, model })
});
return await response.json();
} catch (error) {
console.error('Error analyzing text:', error);
throw error;
}
}
static async getStatistics() {
try {
const response = await fetch('/api/statistics');
return await response.json();
} catch (error) {
console.error('Error getting statistics:', error);
throw error;
}
}
static async getVisualizations() {
try {
const response = await fetch('/api/visualizations');
return await response.json();
} catch (error) {
console.error('Error getting visualizations:', error);
throw error;
}
}
static getSentimentColor(sentiment) {
const colors = {
'Positive': '#28a745',
'Neutral': '#ffc107',
'Negative': '#dc3545'
};
return colors[sentiment] || '#6c757d';
}
static getSentimentIcon(sentiment) {
const icons = {
'Positive': 'fa-smile',
'Neutral': 'fa-meh',
'Negative': 'fa-frown'
};
return icons[sentiment] || 'fa-question-circle';
}
}
// Dashboard initialization
class Dashboard {
constructor() {
this.charts = {};
}
async init() {
try {
const stats = await SentimentAnalyzer.getStatistics();
const viz = await SentimentAnalyzer.getVisualizations();
this.updateStats(stats.statistics);
this.renderCharts(viz.plots);
} catch (error) {
console.error('Error initializing dashboard:', error);
}
}
updateStats(statistics) {
// Update statistics cards
document.querySelectorAll('[data-stat]').forEach(element => {
const statName = element.dataset.stat;
if (statistics[statName] !== undefined) {
if (statName === 'sentiment_distribution') {
// Handle sentiment distribution
const total = Object.values(statistics[statName]).reduce((a, b) => a + b, 0);
element.textContent = total;
} else {
element.textContent = statistics[statName];
}
}
});
}
renderCharts(plots) {
// Render sentiment distribution chart
if (plots.sentiment_distribution) {
const img = document.createElement('img');
img.src = 'data:image/png;base64,' + plots.sentiment_distribution;
img.className = 'img-fluid';
document.getElementById('sentimentChart').appendChild(img);
}
// Render word cloud
if (plots.wordcloud) {
const img = document.createElement('img');
img.src = 'data:image/png;base64,' + plots.wordcloud;
img.className = 'img-fluid';
document.getElementById('wordCloud').appendChild(img);
}
}
}
// Export functionality
class ExportManager {
static async exportToCSV(data, filename = 'sentiment_data.csv') {
const csv = this.convertToCSV(data);
this.downloadFile(csv, filename, 'text/csv');
}
static async exportToJSON(data, filename = 'sentiment_data.json') {
const json = JSON.stringify(data, null, 2);
this.downloadFile(json, filename, 'application/json');
}
static convertToCSV(data) {
if (!Array.isArray(data) || data.length === 0) return '';
const headers = Object.keys(data[0]);
const csvRows = [];
csvRows.push(headers.join(','));
for (const row of data) {
const values = headers.map(header => {
const value = row[header] || '';
return `"${value.toString().replace(/"/g, '""')}"`;
});
csvRows.push(values.join(','));
}
return csvRows.join('\n');
}
static downloadFile(content, filename, contentType) {
const blob = new Blob([content], { type: contentType });
const url = URL.createObjectURL(blob);
const link = document.createElement('a');
link.href = url;
link.download = filename;
document.body.appendChild(link);
link.click();
document.body.removeChild(link);
URL.revokeObjectURL(url);
}
}
// Initialize dashboard when on dashboard page
document.addEventListener('DOMContentLoaded', () => {
if (window.location.pathname === '/dashboard') {
const dashboard = new Dashboard();
dashboard.init();
}
});
// Export button handlers
document.querySelectorAll('[data-export]').forEach(button => {
button.addEventListener('click', async (e) => {
const format = button.dataset.export;
const dataType = button.dataset.data || 'statistics';
try {
let data;
if (dataType === 'statistics') {
const response = await SentimentAnalyzer.getStatistics();
data = response.statistics;
}
if (format === 'csv') {
await ExportManager.exportToCSV([data], `sentiment_${dataType}.csv`);
} else if (format === 'json') {
await ExportManager.exportToJSON(data, `sentiment_${dataType}.json`);
}
} catch (error) {
console.error('Error exporting data:', error);
alert('Error exporting data. Please try again.');
}
});
});

17. .env file

# MongoDB Configuration
MONGODB_URI=mongodb://localhost:27017/
DATABASE_NAME=sentiment_analyzer_db
# Flask Configuration
FLASK_APP=api/app.py
FLASK_ENV=development
SECRET_KEY=your-secret-key-here
# Model Configuration
DEFAULT_MODEL=logistic_regression
MAX_FEATURES=5000
MAX_SEQUENCE_LENGTH=200
# Redis Configuration (for Celery)
REDIS_URL=redis://localhost:6379/0

18. Sample Data File (data/movie_reviews.csv)

review,sentiment
"This movie was absolutely fantastic! Great acting and storyline.",positive
"Terrible movie. Waste of time and money.",negative
"Decent film, but nothing special. Average acting.",neutral
"One of the best movies I've ever seen! Highly recommended.",positive
"Boring and predictable. Fell asleep halfway through.",negative
"Good performances but the plot was confusing.",neutral
"Masterpiece! Director's best work yet.",positive
"Awful script and poor direction. Disappointing.",negative
"Entertaining enough for a Sunday afternoon watch.",neutral
"Brilliant cinematography and emotional depth.",positive

19. run.py (Main entry point)

```python

!/usr/bin/env python3

"""
Sentiment Analyzer - Main entry point
Run this script to start the application
"""

import argparse
import sys
import os
from database.db_operations import MongoDBOperations
from models.sentiment_classifier import SentimentClassifier
from models.deep_learning_model import DeepLearningSentimentClassifier
from models.model_comparator import ModelComparator
from api.app import app
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

def setup_database():
"""Initialize database and load sample data if needed"""
logger.info("Setting up database…")
db_ops = MongoDBOperations()

# Check if database is empty
reviews, _ = db_ops.get_training_data()
if len(reviews) == 0:
logger.info("No training data found. Loading sample data...")
# Load sample movie reviews
sample_data = [
("This movie was absolutely fantastic! Great acting and storyline.", "positive"),
("Terrible movie. Waste of time and money.", "negative"),
("Decent film, but nothing special. Average acting.", "neutral"),
("One of the best movies I've ever seen! Highly recommended.", "positive"),
("Boring and predictable. Fell asleep halfway through.", "negative"),
("Good performances but the plot was confusing.", "neutral"),
("Masterpiece! Director's best work yet.", "positive"),
("Awful script and poor direction. Disappointing.", "negative"),
("Entertaining enough for a Sunday afternoon watch.", "neutral"),
("Brilliant cinematography and emotional depth.", "positive"),
]
for review, sentiment in sample_data:
review_id = db_ops.insert_review(review, source='sample')
db_ops.update_review_label(review_id, sentiment)
logger.info(f"Loaded {len(sample_data)} sample reviews")
db_ops.close_connection()
return True

def train_default_model():
"""Train default model with available data"""
logger.info("Training default model…")

db_ops = MongoDBOperations()
texts, labels = db_ops.get_training_data()
if len(texts) > 0:
# Train traditional ML model
classifier = SentimentClassifier('logistic_regression')
metrics, _, _, _ = classifier.train(texts, labels)
classifier.save_model()
logger.info(f"Model trained with accuracy: {metrics['accuracy']:.4f}")
# Train deep learning model if enough data
if len(texts) >= 100:
dl_classifier = DeepLearningSentimentClassifier('lstm')
dl_metrics, _, _ = dl_classifier.train(texts, labels, epochs=5)
dl_classifier.save_model()
logger.info(f"Deep learning model trained with accuracy: {dl_metrics.get('val_accuracy', 0):.4f}")
db_ops.close_connection()
return True

def run_api(host='0.0.0.0', port=5000, debug=False):
"""Run the Flask API server"""
logger.info(f"Starting API server on {host}:{port}")
app.run(host=host, port=port, debug=debug)

def run_interactive():
"""Run interactive command-line interface"""
from models.sentiment_classifier import SentimentClassifier

print("\n" + "="*60)
print("SENTIMENT ANALYZER - Interactive Mode")
print("="*60)
# Load model
classifier = SentimentClassifier('logistic_regression')
try:
classifier.load_model()
print("✓ Model loaded successfully")
except:
print("! No trained model found. Please train the model first.")
return
while True:
print("\n" + "-"*60)
print("Options:")
print("1. Analyze a review")
print("2. Exit")
choice = input("\nEnter your choice (1-2): ").strip()
if choice == '1':
review = input("\nEnter your review: ").strip()
if review:
print("\nAnalyzing...")
result = classifier.predict(review)
print(f"\nSentiment: {result['sentiment']}")
print(f"Confidence: {result['confidence']:.4f}")
if result.get('probabilities'):
print("\nProbabilities:")
sentiments = ['Negative', 'Neutral', 'Positive']
for sent, prob in zip(sentiments, result['probabilities']):
print(f"  {sent}: {prob:.4f}")
else:
print("Please enter a review")
elif choice == '2':
print("\nGoodbye!")
break
else:
print("Invalid choice. Please try again.")

def main():
parser = argparse.ArgumentParser(description='Sentiment Analyzer Application')
parser.add_argument('--mode', choices=['api', 'train', 'interactive', 'setup'],
default='api', help='

                       default='api', help='Application mode')
parser.add_argument('--host', default='0.0.0.0', help='API host')
parser.add_argument('--port', type=int, default=5000, help='API port')
parser.add_argument('--debug', action='store_true', help='Debug mode')
args = parser.parse_args()
if args.mode == 'setup':
# Setup database and load sample data
setup_database()
train_default_model()
logger.info("Setup completed successfully")
elif args.mode == 'train':
# Train models with existing data
train_default_model()
elif args.mode == 'interactive':
# Run interactive CLI
setup_database()
train_default_model()
run_interactive()
else:  # api mode
# Setup and run API
setup_database()
train_default_model()
run_api(host=args.host, port=args.port, debug=args.debug)
if __name__ == "__main__":
main()

20. tests/test_sentiment_analyzer.py

import unittest
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from models.sentiment_classifier import SentimentClassifier
from models.deep_learning_model import DeepLearningSentimentClassifier
from utils.text_preprocessing import TextPreprocessor
from database.db_operations import MongoDBOperations
import numpy as np
class TestTextPreprocessor(unittest.TestCase):
def setUp(self):
self.preprocessor = TextPreprocessor()
def test_clean_text(self):
text = "This is a TEST!!! with 123 numbers and <html> tags"
cleaned = self.preprocessor.clean_text(text)
self.assertNotIn('123', cleaned)
self.assertNotIn('<html>', cleaned)
self.assertEqual(cleaned, "this is a test with numbers and tags")
def test_convert_emojis(self):
text = "I love this 😊"
converted = self.preprocessor.convert_emojis(text)
self.assertIn('smiling_face', converted)
def test_extract_sentiment_features(self):
text = "This is GREAT!!! 😊😊"
features = self.preprocessor.extract_sentiment_features(text)
self.assertEqual(features['exclamation_count'], 1)
self.assertEqual(features['positive_emoji_count'], 2)
self.assertEqual(features['all_caps_count'], 1)
class TestSentimentClassifier(unittest.TestCase):
def setUp(self):
self.classifier = SentimentClassifier('logistic_regression')
self.test_texts = [
"This movie is fantastic!",
"Terrible movie, waste of time",
"It's an okay film"
]
self.test_labels = ['positive', 'negative', 'neutral']
def test_train_and_predict(self):
# Train model
metrics, _, _, _ = self.classifier.train(self.test_texts, self.test_labels)
self.assertGreater(metrics['accuracy'], 0)
self.assertTrue(self.classifier.is_trained)
# Test prediction
result = self.classifier.predict("Great movie!")
self.assertIn('sentiment', result)
self.assertIn('confidence', result)
def test_predict_batch(self):
self.classifier.train(self.test_texts, self.test_labels)
results = self.classifier.predict_batch(["Good movie", "Bad movie"])
self.assertEqual(len(results), 2)
def test_model_info(self):
self.classifier.train(self.test_texts, self.test_labels)
info = self.classifier.get_model_info()
self.assertTrue(info['is_trained'])
self.assertEqual(info['model_type'], 'logistic_regression')
class TestDeepLearningClassifier(unittest.TestCase):
def setUp(self):
self.classifier = DeepLearningSentimentClassifier('lstm', max_words=1000, max_len=50)
self.test_texts = [
"This movie is fantastic!",
"Terrible movie, waste of time",
"It's an okay film"
]
self.test_labels = ['positive', 'negative', 'neutral']
def test_build_model(self):
self.classifier.build_model()
self.assertIsNotNone(self.classifier.model)
def test_prepare_data(self):
X, y = self.classifier.prepare_data(self.test_texts, self.test_labels)
self.assertEqual(len(X), len(self.test_texts))
self.assertEqual(len(y), len(self.test_labels))
def test_train_and_predict(self):
# Quick test with minimal epochs
metrics, _, _ = self.classifier.train(
self.test_texts, self.test_labels, 
epochs=2, validation_split=0.3
)
self.assertIn('val_accuracy', metrics)
# Test prediction
result = self.classifier.predict("Great movie!")
self.assertIn('sentiment', result)
class TestDatabaseOperations(unittest.TestCase):
def setUp(self):
self.db_ops = MongoDBOperations()
# Use test database
self.db_ops.config.DATABASE_NAME = "test_sentiment_analyzer"
self.db_ops.connect()
def tearDown(self):
# Clean up test database
self.db_ops.client.drop_database("test_sentiment_analyzer")
self.db_ops.close_connection()
def test_insert_review(self):
review_id = self.db_ops.insert_review("Test review", source='test')
self.assertIsNotNone(review_id)
def test_update_review_label(self):
review_id = self.db_ops.insert_review("Test review")
result = self.db_ops.update_review_label(review_id, 'positive')
self.assertTrue(result)
def test_get_training_data(self):
# Insert test data
reviews = ["Good movie", "Bad movie", "Okay movie"]
labels = ["positive", "negative", "neutral"]
for review, label in zip(reviews, labels):
review_id = self.db_ops.insert_review(review)
self.db_ops.update_review_label(review_id, label)
# Retrieve training data
texts, retrieved_labels = self.db_ops.get_training_data()
self.assertEqual(len(texts), 3)
self.assertEqual(len(retrieved_labels), 3)
def test_save_prediction(self):
review_id = self.db_ops.insert_review("Test review")
pred_id = self.db_ops.save_prediction(
review_id, "Test review", "positive", 0.95, 
actual_label="positive"
)
self.assertIsNotNone(pred_id)
class TestModelComparator(unittest.TestCase):
def setUp(self):
from models.model_comparator import ModelComparator
self.comparator = ModelComparator()
self.test_texts = [
"This movie is fantastic!",
"Terrible movie, waste of time",
"It's an okay film",
"Amazing performance!",
"Boring and predictable"
] * 10  # Multiply to get more data
self.test_labels = ['positive', 'negative', 'neutral', 'positive', 'negative'] * 10
def test_initialize_models(self):
self.comparator.initialize_models()
self.assertGreater(len(self.comparator.models), 0)
def test_train_and_compare(self):
self.comparator.initialize_models()
results = self.comparator.train_and_compare(self.test_texts, self.test_labels)
self.assertGreater(len(results), 0)
def test_get_comparison_report(self):
self.comparator.initialize_models()
self.comparator.train_and_compare(self.test_texts, self.test_labels)
df = self.comparator.get_comparison_report()
self.assertFalse(df.empty)
def test_get_best_model(self):
self.comparator.initialize_models()
self.comparator.train_and_compare(self.test_texts, self.test_labels)
best = self.comparator.get_best_model()
self.assertIsNotNone(best)
self.assertIn('model_name', best)
if __name__ == '__main__':
unittest.main()

21. docker-compose.yml

version: '3.8'
services:
mongodb:
image: mongo:6.0
container_name: sentiment-analyzer-mongodb
restart: always
ports:
- "27017:27017"
environment:
MONGO_INITDB_ROOT_USERNAME: admin
MONGO_INITDB_ROOT_PASSWORD: password123
MONGO_INITDB_DATABASE: sentiment_analyzer_db
volumes:
- mongodb_data:/data/db
networks:
- sentiment-network
redis:
image: redis:7-alpine
container_name: sentiment-analyzer-redis
restart: always
ports:
- "6379:6379"
volumes:
- redis_data:/data
networks:
- sentiment-network
api:
build:
context: .
dockerfile: Dockerfile.api
container_name: sentiment-analyzer-api
restart: always
ports:
- "5000:5000"
environment:
MONGODB_URI: mongodb://admin:password123@mongodb:27017/
REDIS_URL: redis://redis:6379/0
FLASK_ENV: production
depends_on:
- mongodb
- redis
volumes:
- ./models/saved_models:/app/models/saved_models
networks:
- sentiment-network
worker:
build:
context: .
dockerfile: Dockerfile.worker
container_name: sentiment-analyzer-worker
restart: always
environment:
MONGODB_URI: mongodb://admin:password123@mongodb:27017/
REDIS_URL: redis://redis:6379/0
depends_on:
- mongodb
- redis
volumes:
- ./models/saved_models:/app/models/saved_models
networks:
- sentiment-network
nginx:
image: nginx:alpine
container_name: sentiment-analyzer-nginx
restart: always
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf
- ./nginx/ssl:/etc/nginx/ssl
depends_on:
- api
networks:
- sentiment-network
volumes:
mongodb_data:
redis_data:
networks:
sentiment-network:
driver: bridge

22. Dockerfile.api

FROM python:3.9-slim
WORKDIR /app
# Install system dependencies
RUN apt-get update && apt-get install -y \
gcc \
g++ \
&& rm -rf /var/lib/apt/lists/*
# Copy requirements
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Download NLTK data
RUN python -c "import nltk; nltk.download('punkt'); nltk.download('stopwords'); nltk.download('wordnet')"
# Copy application code
COPY . .
# Create directory for saved models
RUN mkdir -p models/saved_models
# Expose port
EXPOSE 5000
# Run the application
CMD ["python", "run.py", "--mode", "api", "--host", "0.0.0.0", "--port", "5000"]

23. Dockerfile.worker

FROM python:3.9-slim
WORKDIR /app
# Install system dependencies
RUN apt-get update && apt-get install -y \
gcc \
g++ \
&& rm -rf /var/lib/apt/lists/*
# Copy requirements
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Download NLTK data
RUN python -c "import nltk; nltk.download('punkt'); nltk.download('stopwords'); nltk.download('wordnet')"
# Copy application code
COPY . .
# Create directory for saved models
RUN mkdir -p models/saved_models
# Run worker
CMD ["celery", "-A", "tasks.worker", "worker", "--loglevel=info"]

24. nginx/nginx.conf

events {
worker_connections 1024;
}
http {
upstream api_servers {
server api:5000;
}
server {
listen 80;
server_name localhost;
return 301 https://$server_name$request_uri;
}
server {
listen 443 ssl;
server_name localhost;
ssl_certificate /etc/nginx/ssl/cert.pem;
ssl_certificate_key /etc/nginx/ssl/key.pem;
location / {
proxy_pass http://api_servers;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
location /static {
alias /app/static;
expires 30d;
}
location /api {
proxy_pass http://api_servers;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}
}

25. tasks/worker.py (Celery tasks for background processing)

from celery import Celery
from models.sentiment_classifier import SentimentClassifier
from models.deep_learning_model import DeepLearningSentimentClassifier
from database.db_operations import MongoDBOperations
import logging
logger = logging.getLogger(__name__)
# Initialize Celery
celery = Celery('tasks', broker='redis://localhost:6379/0')
@celery.task
def train_model_task(model_type, texts, labels):
"""Background task to train a model"""
try:
if model_type in ['lstm', 'bilstm']:
model = DeepLearningSentimentClassifier(model_type)
metrics, _, _ = model.train(texts, labels)
model.save_model()
else:
model = SentimentClassifier(model_type)
metrics, _, _, _ = model.train(texts, labels)
model.save_model()
# Save metrics to database
db_ops = MongoDBOperations()
db_ops.update_model_metrics(metrics)
db_ops.close_connection()
return {'success': True, 'metrics': metrics}
except Exception as e:
logger.error(f"Error in training task: {e}")
return {'success': False, 'error': str(e)}
@celery.task
def batch_analyze_task(texts, model_type='logistic_regression'):
"""Background task to analyze multiple texts"""
try:
if model_type in ['lstm', 'bilstm']:
model = DeepLearningSentimentClassifier(model_type)
model.load_model()
results = model.predict_batch(texts)
else:
model = SentimentClassifier(model_type)
model.load_model()
results = model.predict_batch(texts)
# Save to database
db_ops = MongoDBOperations()
for text, result in zip(texts, results):
review_id = db_ops.insert_review(text)
db_ops.save_prediction(
review_id, text, result['sentiment'],
result['confidence'], model_used=model_type
)
db_ops.close_connection()
return {'success': True, 'results': results}
except Exception as e:
logger.error(f"Error in batch analysis: {e}")
return {'success': False, 'error': str(e)}
@celery.task
def export_data_task(format='csv'):
"""Background task to export data"""
try:
db_ops = MongoDBOperations()
df = db_ops.export_reviews_to_dataframe()
db_ops.close_connection()
if format == 'csv':
data = df.to_csv(index=False)
else:
data = df.to_json(orient='records')
return {'success': True, 'data': data, 'format': format}
except Exception as e:
logger.error(f"Error in export task: {e}")
return {'success': False, 'error': str(e)}

26. scripts/generate_test_data.py

#!/usr/bin/env python3
"""
Script to generate test data for sentiment analysis
"""
import random
import csv
import argparse
def generate_movie_reviews(num_reviews=1000):
"""Generate synthetic movie reviews"""
positive_templates = [
"This movie was absolutely {}!",
"{} performance by the lead actor.",
"One of the best {} movies I've seen.",
"Absolutely {} from start to finish.",
"A {} masterpiece of cinema.",
"The {} direction made this film special.",
"{} acting and {} storyline.",
"I was {} throughout the entire movie.",
"This film {} exceeded my expectations.",
"A {} gem that deserves more attention."
]
negative_templates = [
"This movie was completely {}.",
"{} acting ruined the experience.",
"One of the worst {} movies ever.",
"Absolutely {} and {}.",
"A {} disaster from start to finish.",
"The {} direction made no sense.",
"{} plot and {} characters.",
"I was {} bored throughout.",
"This film {} disappointed me.",
"A {} waste of time and money."
]
neutral_templates = [
"This movie was {}.",
"{} performance from the cast.",
"An {} film with some issues.",
"{} entertaining but {}.",
"The movie had its {} moments.",
"Nothing {} about this film.",
"{} average movie overall.",
"I had {} expectations for this.",
"The film was {} but {}.",
"A {} viewing experience."
]
positive_adjectives = ['fantastic', 'amazing', 'brilliant', 'excellent', 'wonderful',
'superb', 'outstanding', 'remarkable', 'incredible', 'spectacular']
negative_adjectives = ['terrible', 'awful', 'horrible', 'dreadful', 'pathetic',
'disappointing', 'mediocre', 'poor', 'boring', 'ridiculous']
neutral_adjectives = ['okay', 'decent', 'average', 'mediocre', 'fair',
'moderate', 'reasonable', 'acceptable', 'passable', 'tolerable']
reviews = []
for _ in range(num_reviews):
sentiment = random.choice(['positive', 'negative', 'neutral'])
if sentiment == 'positive':
template = random.choice(positive_templates)
adj1 = random.choice(positive_adjectives)
adj2 = random.choice(positive_adjectives)
review = template.format(adj1, adj2)
elif sentiment == 'negative':
template = random.choice(negative_templates)
adj1 = random.choice(negative_adjectives)
adj2 = random.choice(negative_adjectives)
review = template.format(adj1, adj2)
else:  # neutral
template = random.choice(neutral_templates)
adj1 = random.choice(neutral_adjectives)
adj2 = random.choice(neutral_adjectives)
review = template.format(adj1, adj2)
reviews.append((review, sentiment))
return reviews
def generate_product_reviews(num_reviews=1000):
"""Generate synthetic product reviews"""
positive_templates = [
"This product is {}!",
"{} quality and {} design.",
"Best {} I've ever purchased.",
"Absolutely {} value for money.",
"{} product, highly recommended.",
"The {} features are impressive.",
"{} build quality and {} performance.",
"I'm {} satisfied with this purchase.",
"This {} exceeded my expectations.",
"A {} product that delivers."
]
negative_templates = [
"This product is completely {}.",
"{} quality and {} design.",
"Worst {} I've ever purchased.",
"Absolutely {} waste of money.",
"{} product, not recommended.",
"The {} features are useless.",
"{} build quality and {} performance.",
"I'm {} disappointed with this.",
"This {} failed to meet expectations.",
"A {} product that doesn't work."
]
neutral_templates = [
"This product is {}.",
"{} quality but {} design.",
"An {} product with some issues.",
"{} value for money.",
"The product is {} overall.",
"It has some {} features.",
"{} build quality and {} performance.",
"I'm {} about this purchase.",
"This product is {} but {}.",
"A {} product that's {}."
]
# Reuse adjectives from movie reviews
positive_adjectives = ['fantastic', 'amazing', 'brilliant', 'excellent', 'wonderful',
'superb', 'outstanding', 'remarkable', 'incredible', 'great']
negative_adjectives = ['terrible', 'awful', 'horrible', 'dreadful', 'poor',
'disappointing', 'mediocre', 'cheap', 'flimsy', 'broken']
neutral_adjectives = ['okay', 'decent', 'average', 'mediocre', 'fair',
'reasonable', 'acceptable', 'passable', 'functional', 'standard']
reviews = []
for _ in range(num_reviews):
sentiment = random.choice(['positive', 'negative', 'neutral'])
if sentiment == 'positive':
template = random.choice(positive_templates)
adj1 = random.choice(positive_adjectives)
adj2 = random.choice(positive_adjectives)
review = template.format(adj1, adj2)
elif sentiment == 'negative':
template = random.choice(negative_templates)
adj1 = random.choice(negative_adjectives)
adj2 = random.choice(negative_adjectives)
review = template.format(adj1, adj2)
else:  # neutral
template = random.choice(neutral_templates)
adj1 = random.choice(neutral_adjectives)
adj2 = random.choice(neutral_adjectives)
review = template.format(adj1, adj2)
reviews.append((review, sentiment))
return reviews
def save_to_csv(reviews, filename):
"""Save reviews to CSV file"""
with open(filename, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(['review', 'sentiment'])
writer.writerows(reviews)
print(f"Saved {len(reviews)} reviews to {filename}")
def main():
parser = argparse.ArgumentParser(description='Generate test data for sentiment analysis')
parser.add_argument('--type', choices=['movie', 'product', 'both'], default='both',
help='Type of reviews to generate')
parser.add_argument('--num', type=int, default=1000,
help='Number of reviews to generate per type')
parser.add_argument('--output', default='data/',
help='Output directory')
args = parser.parse_args()
if args.type in ['movie', 'both']:
print(f"Generating {args.num} movie reviews...")
movie_reviews = generate_movie_reviews(args.num)
save_to_csv(movie_reviews, f"{args.output}/movie_reviews.csv")
if args.type in ['product', 'both']:
print(f"Generating {args.num} product reviews...")
product_reviews = generate_product_reviews(args.num)
save_to_csv(product_reviews, f"{args.output}/product_reviews.csv")
print("Done!")
if __name__ == "__main__":
main()

27. README.md

# Sentiment Analyzer - Movie & Product Review Analysis
## 📋 Overview
A comprehensive sentiment analysis system that classifies movie and product reviews into Positive, Negative, or Neutral categories using multiple machine learning and deep learning models.
## ✨ Features
- **Multi-model Support**: Naive Bayes, Logistic Regression, SVM, Random Forest, LSTM, Bi-LSTM
- **Real-time Analysis**: Instant sentiment prediction with confidence scores
- **Batch Processing**: Analyze multiple reviews simultaneously
- **Interactive Dashboard**: Visualize sentiment distributions and model performance
- **RESTful API**: Easy integration with other applications
- **MongoDB Integration**: Persistent storage of reviews and predictions
- **Model Comparison**: Compare performance across different algorithms
- **Export Functionality**: Export results to CSV, JSON, or Excel
- **Docker Support**: Easy deployment with Docker Compose
## 🚀 Quick Start
### Prerequisites
- Python 3.8+
- MongoDB 4.0+
- Redis (optional, for background tasks)
- Docker (optional)
### Installation
1. **Clone the repository**

bash
git clone
cd sentiment-analyzer

2. **Install dependencies**

bash
pip install -r requirements.txt

3. **Set up environment variables**

bash
cp .env.example .env

Edit .env with your configuration

4. **Start MongoDB**

bash

Using Docker

docker run -d -p 27017:27017 --name mongodb mongo:6.0

Or local installation

sudo systemctl start mongodb

5. **Run the application**

bash

Setup database and train models

python run.py --mode setup

Start API server

python run.py --mode api

Or run interactive CLI

python run.py --mode interactive

### Using Docker Compose

bash
docker-compose up -d

## 📊 Usage Examples
### Python API

python
from models.sentiment_classifier import SentimentClassifier

Initialize classifier

classifier = SentimentClassifier('logistic_regression')
classifier.load_model()

Analyze single review

result = classifier.predict("This movie was absolutely fantastic!")
print(f"Sentiment: {result['sentiment']}")
print(f"Confidence: {result['confidence']}")

Batch analysis

reviews = ["Great movie!", "Terrible film", "It was okay"]
results = classifier.predict_batch(reviews)

### REST API

bash

Analyze sentiment

curl -X POST http://localhost:5000/api/analyze \
-H "Content-Type: application/json" \
-d '{"text": "This movie is amazing!", "model": "logistic_regression"}'

Batch analysis

curl -X POST http://localhost:5000/api/analyze/batch \
-H "Content-Type: application/json" \
-d '{"texts": ["Great!", "Bad!"], "model": "lstm"}'

Get statistics

curl http://localhost:5000/api/statistics

### Web Interface
- Open http://localhost:5000 in your browser
- Navigate to Analyze page to test individual reviews
- View Dashboard for statistics and visualizations
## 🏗️ Project Structure

sentiment-analyzer/
├── api/ # Flask API routes
├── config/ # Configuration files
├── data/ # Dataset files
├── database/ # MongoDB operations
├── models/ # ML/DL models
├── nginx/ # Nginx configuration
├── scripts/ # Utility scripts
├── tasks/ # Celery tasks
├── tests/ # Unit tests
├── utils/ # Helper functions
├── web/ # Web interface
├── docker-compose.yml
├── requirements.txt
└── run.py

## 🤖 Models
### Traditional Machine Learning
- **Naive Bayes**: Fast and efficient for text classification
- **Logistic Regression**: Good baseline with interpretable results
- **SVM**: Effective for high-dimensional spaces
- **Random Forest**: Ensemble method for robust predictions
### Deep Learning
- **LSTM**: Captures long-term dependencies in text
- **Bi-LSTM**: Bidirectional context understanding
- **CNN-LSTM**: Hybrid architecture for local patterns + sequences
## 📈 Performance Metrics
- **Accuracy**: Overall correctness
- **Precision**: Positive prediction accuracy
- **Recall**: Coverage of actual positives
- **F1-Score**: Harmonic mean of precision and recall
- **Confusion Matrix**: Detailed classification breakdown
## 🔧 Configuration
### Model Settings (config/model_config.py)

python
MAX_FEATURES = 5000 # Maximum vocabulary size
MAX_SEQUENCE_LENGTH = 200 # Maximum text length for DL models
BATCH_SIZE = 32 # Training batch size
EPOCHS = 10 # Number of training epochs

### Database Settings (config/mongodb_config.py)

python
MONGODB_URI = "mongodb://localhost:27017/"
DATABASE_NAME = "sentiment_analyzer_db"

## 📁 Data Format
### Training Data CSV

csv
review,sentiment
"This movie is fantastic!",positive
"Terrible product, waste of money",negative
"It's an okay film",neutral

## 🧪 Testing

bash

Run all tests

python -m unittest discover tests

Run specific test

python -m unittest tests.test_sentiment_analyzer.TestSentimentClassifier

## 📦 Deployment
### Production Deployment with Docker

bash

Build and start all services

docker-compose up -d --build

Scale API servers

docker-compose up -d --scale api=3

View logs

docker-compose logs -f

### Manual Deployment

bash

Using Gunicorn

gunicorn -w 4 -b 0.0.0.0:5000 api.app:app

With Nginx as reverse proxy

(See nginx/nginx.conf for configuration)

## 🔒 Security Considerations
- Input validation and sanitization
- Rate limiting on API endpoints
- MongoDB authentication enabled
- HTTPS with SSL/TLS
- Environment variables for sensitive data
## 📊 Monitoring
- Prometheus metrics endpoint
- Grafana dashboards
- ELK stack integration
- Sentry for error tracking
## 🤝 Contributing
1. Fork the repository
2. Create a feature branch
3. Commit your changes
4. Push to the branch
5. Open a Pull Request
## 📄 License
MIT License - see LICENSE file for details
## 👥 Authors
- Your Name - Initial work
## 🙏 Acknowledgments
- NLTK for text processing utilities
- Scikit-learn for ML algorithms
- TensorFlow/Keras for deep learning
- MongoDB for database storage
## 📞 Support
For issues and questions:
- Create an issue in the repository
- Contact: [email protected]
## 🚀 Future Enhancements
- [ ] BERT/RoBERTa transformer models
- [ ] Multi-language support
- [ ] Aspect-based sentiment analysis
- [ ] Real-time streaming analysis
- [ ] Active learning for model improvement
- [ ] A/B testing framework
- [ ] Model versioning and rollback
- [ ] Automated retraining pipeline

This completes the comprehensive Sentiment Analyzer project! The system includes:

1. Multiple ML/DL Models: Traditional ML (Naive Bayes, Logistic Regression, SVM, Random Forest) and Deep Learning (LSTM, Bi-LSTM)
2. Full-stack Implementation: Backend API, Web Interface, Database Integration
3. Production-ready Features: Docker support, Nginx configuration, Celery tasks
4. Comprehensive Testing: Unit tests for all components
5. Documentation: Detailed README with setup instructions
6. Scalability: Designed to handle large volumes of reviews
7. Extensibility: Easy to add new models and features

The system can be deployed in various environments and scaled horizontally using Docker containers. It provides both REST API for integration and a user-friendly web interface for interactive use.