""" NLP Text Classification Pipeline Sentiment Analysis on Twitter/Social Media Data """ import pandas as pd import numpy as np import re import string import warnings warnings.filterwarnings('ignore') # NLTK imports import nltk from nltk.corpus import stopwords from nltk.tokenize import word_tokenize from nltk.stem import WordNetLemmatizer, PorterStemmer # Sklearn imports from sklearn.model_selection import train_test_split from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer from sklearn.naive_bayes import MultinomialNB from sklearn.linear_model import LogisticRegression from sklearn.svm import SVC from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import ( accuracy_score, precision_score, recall_score, f1_score, classification_report, confusion_matrix ) # Download required NLTK data def download_nltk_data(): """Download required NLTK resources""" resources = ['punkt', 'stopwords', 'wordnet', 'punkt_tab'] for resource in resources: try: nltk.download(resource, quiet=True) except: pass download_nltk_data() # ============================================================================= # 1. DATA LOADING # ============================================================================= def load_dataset(filepath): """Load dataset from CSV file""" print(f"Loading dataset from: {filepath}") df = pd.read_csv(filepath) print(f"Dataset shape: {df.shape}") print(f"Columns: {df.columns.tolist()}") return df def explore_dataset(df): """Explore and display dataset statistics""" print("\n" + "="*50) print("DATASET EXPLORATION") print("="*50) print(f"\nDataset Shape: {df.shape}") print(f"\nColumn Names: {df.columns.tolist()}") print(f"\nData Types:\n{df.dtypes}") print(f"\nMissing Values:\n{df.isnull().sum()}") # Find text and label columns text_col = None label_col = None for col in df.columns: if 'text' in col.lower(): text_col = col if 'sentiment' in col.lower() or 'label' in col.lower(): label_col = col if label_col: print(f"\nClass Distribution:") print(df[label_col].value_counts()) print(f"\nClass Percentages:") print(df[label_col].value_counts(normalize=True) * 100) print(f"\nSample Data:") print(df.head()) return text_col, label_col # ============================================================================= # 2. DATA PREPROCESSING # ============================================================================= class TextPreprocessor: """Text preprocessing class for NLP tasks""" def __init__(self, remove_stopwords=True, lemmatize=True, stem=False): self.remove_stopwords = remove_stopwords self.lemmatize = lemmatize self.stem = stem self.stop_words = set(stopwords.words('english')) self.lemmatizer = WordNetLemmatizer() self.stemmer = PorterStemmer() def clean_text(self, text): """Clean and preprocess a single text""" if pd.isna(text): return "" text = str(text) # Convert to lowercase text = text.lower() # Remove URLs text = re.sub(r'http\S+|www\S+|https\S+', '', text, flags=re.MULTILINE) # Remove user mentions text = re.sub(r'@\w+', '', text) # Remove hashtags text = re.sub(r'#\w+', '', text) # Remove HTML tags text = re.sub(r'<.*?>', '', text) # Remove special characters and numbers text = re.sub(r'[^a-zA-Z\s]', '', text) # Remove extra whitespace text = ' '.join(text.split()) return text def tokenize_and_process(self, text): """Tokenize and apply stemming/lemmatization""" try: tokens = word_tokenize(text) except: tokens = text.split() # Remove stopwords if self.remove_stopwords: tokens = [t for t in tokens if t not in self.stop_words] # Apply lemmatization or stemming if self.lemmatize: tokens = [self.lemmatizer.lemmatize(t) for t in tokens] elif self.stem: tokens = [self.stemmer.stem(t) for t in tokens] return ' '.join(tokens) def preprocess(self, text): """Full preprocessing pipeline""" text = self.clean_text(text) text = self.tokenize_and_process(text) return text def preprocess_dataframe(self, df, text_column): """Preprocess entire dataframe""" print("Preprocessing text data...") df['processed_text'] = df[text_column].apply(self.preprocess) print("Preprocessing complete!") return df # ============================================================================= # 3. DATA SPLITTING # ============================================================================= def split_dataset(X, y, test_size=0.2, val_size=0.1, random_state=42): """ Split dataset into train, validation, and test sets Parameters: - X: Features (text data) - y: Labels - test_size: Proportion of test set - val_size: Proportion of validation set (from remaining after test split) - random_state: Random seed for reproducibility """ print("\n" + "="*50) print("SPLITTING DATASET") print("="*50) # First split: separate test set X_temp, X_test, y_temp, y_test = train_test_split( X, y, test_size=test_size, random_state=random_state, stratify=y ) # Second split: separate validation from training if val_size > 0: val_ratio = val_size / (1 - test_size) X_train, X_val, y_train, y_val = train_test_split( X_temp, y_temp, test_size=val_ratio, random_state=random_state, stratify=y_temp ) print(f"Training set size: {len(X_train)}") print(f"Validation set size: {len(X_val)}") print(f"Test set size: {len(X_test)}") return X_train, X_val, X_test, y_train, y_val, y_test else: print(f"Training set size: {len(X_temp)}") print(f"Test set size: {len(X_test)}") return X_temp, X_test, y_temp, y_test # ============================================================================= # 4. FEATURE REPRESENTATION # ============================================================================= class FeatureExtractor: """Feature extraction methods for text data""" def __init__(self, method='tfidf', max_features=5000, ngram_range=(1, 2)): self.method = method self.max_features = max_features self.ngram_range = ngram_range self.vectorizer = None def fit_transform(self, X_train): """Fit vectorizer on training data and transform""" print(f"\nApplying {self.method.upper()} feature extraction...") if self.method == 'tfidf': self.vectorizer = TfidfVectorizer( max_features=self.max_features, ngram_range=self.ngram_range, min_df=2, max_df=0.95 ) elif self.method == 'bow': self.vectorizer = CountVectorizer( max_features=self.max_features, ngram_range=self.ngram_range, min_df=2, max_df=0.95 ) else: raise ValueError(f"Unknown method: {self.method}") X_transformed = self.vectorizer.fit_transform(X_train) print(f"Feature matrix shape: {X_transformed.shape}") return X_transformed def transform(self, X): """Transform new data using fitted vectorizer""" if self.vectorizer is None: raise ValueError("Vectorizer not fitted. Call fit_transform first.") return self.vectorizer.transform(X) def get_feature_names(self): """Get feature names from vectorizer""" return self.vectorizer.get_feature_names_out() # ============================================================================= # 5. MODEL TRAINING # ============================================================================= class ClassifierTrainer: """Train and compare multiple classifiers""" def __init__(self): self.models = { 'Naive Bayes': MultinomialNB(), 'Logistic Regression': LogisticRegression(max_iter=1000, random_state=42), 'SVM': SVC(kernel='linear', random_state=42), 'Random Forest': RandomForestClassifier(n_estimators=100, random_state=42, n_jobs=-1) } self.trained_models = {} self.results = {} def train_model(self, model_name, X_train, y_train): """Train a single model""" print(f"\nTraining {model_name}...") model = self.models[model_name] model.fit(X_train, y_train) self.trained_models[model_name] = model print(f"{model_name} training complete!") return model def train_all(self, X_train, y_train): """Train all models""" print("\n" + "="*50) print("TRAINING MODELS") print("="*50) for name in self.models: self.train_model(name, X_train, y_train) return self.trained_models # ============================================================================= # 6. MODEL EVALUATION # ============================================================================= class ModelEvaluator: """Evaluate and compare model performance""" def __init__(self, label_names=None): self.label_names = label_names self.results = [] def evaluate_model(self, model, model_name, X_test, y_test): """Evaluate a single model""" y_pred = model.predict(X_test) accuracy = accuracy_score(y_test, y_pred) precision = precision_score(y_test, y_pred, average='weighted') recall = recall_score(y_test, y_pred, average='weighted') f1 = f1_score(y_test, y_pred, average='weighted') result = { 'Model': model_name, 'Accuracy': accuracy, 'Precision': precision, 'Recall': recall, 'F1-Score': f1 } self.results.append(result) return result, y_pred def evaluate_all(self, trained_models, X_test, y_test): """Evaluate all trained models""" print("\n" + "="*50) print("MODEL EVALUATION") print("="*50) for name, model in trained_models.items(): result, y_pred = self.evaluate_model(model, name, X_test, y_test) print(f"\n{'-'*40}") print(f"Model: {name}") print(f"{'-'*40}") print(f"Accuracy: {result['Accuracy']:.4f}") print(f"Precision: {result['Precision']:.4f}") print(f"Recall: {result['Recall']:.4f}") print(f"F1-Score: {result['F1-Score']:.4f}") print(f"\nClassification Report:") print(classification_report(y_test, y_pred)) print(f"Confusion Matrix:") print(confusion_matrix(y_test, y_pred)) return self.results def get_results_dataframe(self): """Return results as DataFrame""" return pd.DataFrame(self.results) def get_best_model(self, trained_models, metric='F1-Score'): """Get the best performing model based on specified metric""" results_df = self.get_results_dataframe() best_idx = results_df[metric].idxmax() best_model_name = results_df.loc[best_idx, 'Model'] return best_model_name, trained_models[best_model_name] # ============================================================================= # 7. MAIN PIPELINE # ============================================================================= def run_pipeline(filepath, use_preprocessed=False): """ Run the complete NLP classification pipeline Parameters: - filepath: Path to the CSV file - use_preprocessed: If True, assumes text is already preprocessed """ print("\n" + "="*60) print("NLP TEXT CLASSIFICATION PIPELINE") print("="*60) # 1. Load data df = load_dataset(filepath) text_col, label_col = explore_dataset(df) # Determine text column if 'clean_text' in df.columns: text_col = 'clean_text' use_preprocessed = True elif 'text' in df.columns: text_col = 'text' elif 'processed_text' in df.columns: text_col = 'processed_text' use_preprocessed = True print(f"\nUsing text column: {text_col}") print(f"Using label column: {label_col}") # 2. Preprocess if needed if not use_preprocessed: preprocessor = TextPreprocessor( remove_stopwords=True, lemmatize=True, stem=False ) df = preprocessor.preprocess_dataframe(df, text_col) text_col = 'processed_text' # Remove empty texts df = df[df[text_col].str.len() > 0] # Encode labels from sklearn.preprocessing import LabelEncoder le = LabelEncoder() df['label_encoded'] = le.fit_transform(df[label_col]) print(f"\nLabel mapping: {dict(zip(le.classes_, le.transform(le.classes_)))}") # 3. Split data X = df[text_col].values y = df['label_encoded'].values X_train, X_test, y_train, y_test = split_dataset(X, y, test_size=0.2, val_size=0) # 4. Feature extraction - try both methods results_all = [] for method in ['tfidf', 'bow']: print(f"\n{'='*60}") print(f"USING {method.upper()} FEATURES") print(f"{'='*60}") feature_extractor = FeatureExtractor(method=method, max_features=5000) X_train_features = feature_extractor.fit_transform(X_train) X_test_features = feature_extractor.transform(X_test) # 5. Train models trainer = ClassifierTrainer() trained_models = trainer.train_all(X_train_features, y_train) # 6. Evaluate models evaluator = ModelEvaluator(label_names=le.classes_) evaluator.evaluate_all(trained_models, X_test_features, y_test) results_df = evaluator.get_results_dataframe() results_df['Feature Method'] = method.upper() results_all.append(results_df) # Get best model best_name, best_model = evaluator.get_best_model(trained_models) print(f"\nBest model with {method.upper()}: {best_name}") # Combine all results final_results = pd.concat(results_all, ignore_index=True) print("\n" + "="*60) print("FINAL RESULTS COMPARISON") print("="*60) print(final_results.to_string(index=False)) # Find overall best best_idx = final_results['F1-Score'].idxmax() print(f"\n{'='*60}") print(f"BEST OVERALL MODEL:") print(f"{'='*60}") print(f"Model: {final_results.loc[best_idx, 'Model']}") print(f"Feature Method: {final_results.loc[best_idx, 'Feature Method']}") print(f"F1-Score: {final_results.loc[best_idx, 'F1-Score']:.4f}") return final_results # ============================================================================= # MAIN EXECUTION # ============================================================================= if __name__ == "__main__": # Run pipeline on both datasets print("\n" + "#"*70) print("# RUNNING PIPELINE ON CLEAN DATASET") print("#"*70) results_clean = run_pipeline('clean_dataset.csv', use_preprocessed=True) print("\n\n" + "#"*70) print("# RUNNING PIPELINE ON UNBALANCED DATASET") print("#"*70) results_unbalanced = run_pipeline('unbalanceddataset.csv', use_preprocessed=False) # Save results to CSV results_clean.to_csv('results_clean_dataset.csv', index=False) results_unbalanced.to_csv('results_unbalanced_dataset.csv', index=False) print("\n\nResults saved to CSV files!")