""" preprocessing_utils.py ====================== Shared NLP preprocessing utilities for all model pipelines. Handles text cleaning, normalization, and sequence preparation. """ import re import string import numpy as np from typing import List, Optional # ─── NLTK Setup ────────────────────────────────────────────────────────────── def ensure_nltk_resources(): """Download required NLTK data if not already present.""" import nltk resources = [ ("corpora/stopwords", "stopwords"), ("tokenizers/punkt", "punkt"), ("corpora/wordnet", "wordnet"), ] for path, name in resources: try: nltk.data.find(path) except LookupError: nltk.download(name, quiet=True) ensure_nltk_resources() import nltk from nltk.corpus import stopwords from nltk.stem import WordNetLemmatizer STOP_WORDS = set(stopwords.words("english")) # Preserve negation words — removing them destroys semantic inversion signals _NEGATION_WORDS = { "no", "nor", "not", "never", "don't", "didn't", "isn't", "wasn't", "aren't", "won't", "can't", "couldn't", "shouldn't", "wouldn't", "hasn't", "haven't", "hadn't", "doesn't", "n't", } STOP_WORDS = STOP_WORDS - _NEGATION_WORDS LEMMATIZER = WordNetLemmatizer() # ─── Contraction Expansion ──────────────────────────────────────────────────── CONTRACTION_MAP = { "don't": "do not", "didn't": "did not", "doesn't": "does not", "won't": "will not", "wouldn't": "would not", "couldn't": "could not", "shouldn't": "should not", "can't": "cannot", "couldn't": "could not", "isn't": "is not", "aren't": "are not", "wasn't": "was not", "weren't": "were not", "hasn't": "has not", "haven't": "have not", "hadn't": "had not", "ain't": "is not", "i'm": "i am", "you're": "you are", "he's": "he is", "she's": "she is", "it's": "it is", "we're": "we are", "they're": "they are", "i've": "i have", "you've": "you have", "we've": "we have", "they've": "they have", "i'll": "i will", "you'll": "you will", "he'll": "he will", "she'll": "she will", "we'll": "we will", "they'll": "they will", "i'd": "i would", "you'd": "you would", "he'd": "he would", "she'd": "she would", "we'd": "we would", "they'd": "they would", "let's": "let us", } _CONTRACTION_PATTERN = re.compile( r"\b(" + "|".join(re.escape(k) for k in CONTRACTION_MAP) + r")\b", flags=re.IGNORECASE, ) def expand_contractions(text: str) -> str: """Expand English contractions to their full forms.""" def _replace(m: re.Match) -> str: key = m.group(1).lower() return CONTRACTION_MAP.get(key, m.group(1)) return _CONTRACTION_PATTERN.sub(_replace, text) # ─── Core Cleaning ─────────────────────────────────────────────────────────── def remove_html_tags(text: str) -> str: """Strip HTML tags from text.""" return re.sub(r"<[^>]+>", " ", text) def remove_urls(text: str) -> str: """Remove URLs from text.""" return re.sub(r"http\S+|www\S+|https\S+", " ", text, flags=re.MULTILINE) def remove_special_characters(text: str) -> str: """Remove non-alphabetical characters, keep spaces.""" return re.sub(r"[^a-zA-Z\s]", " ", text) def normalize_whitespace(text: str) -> str: """Collapse multiple spaces into one.""" return re.sub(r"\s+", " ", text).strip() # ─── Traditional ML Preprocessing (TF-IDF) ──────────────────────────────── def preprocess_for_tfidf(text: str, remove_stopwords: bool = True) -> str: """ Full preprocessing pipeline for TF-IDF / Logistic Regression. Steps: 1. Lowercase 2. Expand contractions (e.g. didn't -> did not) 3. Remove HTML 4. Remove URLs 5. Remove special chars 6. Optional stopword removal 7. Lemmatize 8. Normalize whitespace """ text = text.lower() text = expand_contractions(text) text = remove_html_tags(text) text = remove_urls(text) text = remove_special_characters(text) tokens = text.split() if remove_stopwords: tokens = [t for t in tokens if t not in STOP_WORDS] tokens = [LEMMATIZER.lemmatize(t) for t in tokens if len(t) > 1] return normalize_whitespace(" ".join(tokens)) # ─── Deep Learning Preprocessing (LSTM) ────────────────────────────────── def preprocess_for_lstm(text: str) -> str: """ Preprocessing pipeline for LSTM/sequence models. Matches clean_text in retrain_lstm.py exactly to align train and inference. Steps: 1. Lowercase 2. Expand contractions 3. Remove HTML 4. Remove URLs 5. Keep alphanumeric and basic punctuation (!?.,) 6. Normalize whitespace """ text = text.lower() text = expand_contractions(text) text = remove_html_tags(text) text = remove_urls(text) text = re.sub(r'[^a-zA-Z0-9!?., ]', '', text) return normalize_whitespace(text) def text_to_sequence(text: str, tokenizer, max_len: int = 300) -> np.ndarray: """ Convert text to padded integer sequence for LSTM inference. Args: text: Raw or cleaned text string. tokenizer: Fitted Keras Tokenizer object. max_len: Maximum sequence length (pad/truncate to this). Returns: Numpy array of shape (1, max_len). """ from tensorflow.keras.preprocessing.sequence import pad_sequences # type: ignore cleaned = preprocess_for_lstm(text) seq = tokenizer.texts_to_sequences([cleaned]) padded = pad_sequences(seq, maxlen=max_len, padding="post", truncating="post") return padded # ─── Transformer Preprocessing (BERT) ──────────────────────────────────── def preprocess_for_bert(text: str) -> str: """ Minimal preprocessing for BERT — the tokenizer handles most normalization. BERT's WordPiece tokenizer is robust; we only: 1. Remove HTML tags 2. Collapse excessive whitespace """ text = remove_html_tags(text) return normalize_whitespace(text) # ─── Text Analysis Utilities ────────────────────────────────────────────── def get_word_count(text: str) -> int: """Count words in text.""" return len(text.split()) def get_char_count(text: str) -> int: """Count characters (excluding spaces) in text.""" return len(text.replace(" ", "")) def detect_negation(text: str) -> bool: """Simple negation detection for reasoning engine.""" negation_words = { "not", "no", "never", "none", "nobody", "nothing", "neither", "nowhere", "nor", "cannot", "can't", "won't", "don't", "doesn't", "didn't", "isn't", "aren't", "wasn't", "weren't", "hasn't", "haven't", "hadn't", "wouldn't", "shouldn't", "couldn't", "n't", "hardly", "barely", "scarcely" } tokens = set(text.lower().split()) return bool(tokens & negation_words) def detect_sarcasm_signals(text: str) -> bool: """Heuristic sarcasm/irony signal detection.""" sarcasm_patterns = [ r"\b(yeah right|sure sure|totally|oh great|fantastic job|brilliant)\b", r"(!!!|\?\?\?)", r"\b(not really|kind of|sort of)\b", r"(worst.*best|best.*worst)", ] text_lower = text.lower() return any(re.search(p, text_lower) for p in sarcasm_patterns) def detect_mixed_sentiment(text: str) -> bool: """Detect presence of both positive and negative signals.""" positive_words = { "great", "good", "excellent", "amazing", "wonderful", "fantastic", "love", "loved", "enjoy", "enjoyed", "brilliant", "superb", "perfect" } negative_words = { "bad", "terrible", "awful", "horrible", "boring", "hate", "hated", "disappointing", "disappointed", "poor", "worst", "dull", "weak" } tokens = set(text.lower().split()) has_pos = bool(tokens & positive_words) has_neg = bool(tokens & negative_words) return has_pos and has_neg def extract_key_phrases(text: str, top_n: int = 5) -> List[str]: """ Extract simple keyword phrases for reasoning display. Returns tokens most likely to influence sentiment. """ sentiment_vocab = { # Strong positive "excellent", "amazing", "outstanding", "brilliant", "masterpiece", "fantastic", "wonderful", "superb", "love", "perfect", "beautiful", "great", "good", "enjoy", "entertaining", "captivating", "thrilling", # Strong negative "terrible", "awful", "horrible", "boring", "waste", "disappointing", "bad", "poor", "worst", "dreadful", "pathetic", "ridiculous", "unbearable", "painful", "disaster", "failure", "mediocre", "weak", # Modifiers "very", "extremely", "absolutely", "completely", "totally", "quite", "never", "not", "no", "hardly", "barely", } tokens = re.findall(r"\b[a-zA-Z]+\b", text.lower()) found = [t for t in tokens if t in sentiment_vocab] # Deduplicate while preserving order seen = set() unique = [] for t in found: if t not in seen: seen.add(t) unique.append(t) return unique[:top_n] def generate_reasoning( text: str, model_name: str, prediction: str, confidence: float, ) -> str: """ Generate a human-readable reasoning explanation for a prediction. Args: text: Input review text. model_name: One of 'Logistic Regression', 'Bi-LSTM', 'BERT'. prediction: 'Positive' or 'Negative'. confidence: Confidence score 0–1. Returns: A natural-language reasoning string. """ has_negation = detect_negation(text) has_sarcasm = detect_sarcasm_signals(text) has_mixed = detect_mixed_sentiment(text) key_phrases = extract_key_phrases(text) word_count = get_word_count(text) phrases_str = ", ".join([f'"{p}"' for p in key_phrases]) if key_phrases else "the overall tone" conf_label = "high" if confidence >= 0.80 else "moderate" if confidence >= 0.60 else "low" reasoning_parts = [] # Model-specific reasoning if model_name == "Logistic Regression": reasoning_parts.append( f"TF-IDF features weighted {phrases_str} as the primary sentiment signals." ) if has_negation and confidence < 0.75: reasoning_parts.append( "Negation patterns may have reduced confidence — TF-IDF treats tokens independently." ) reasoning_parts.append( f"Bag-of-words representation captured {conf_label} confidence based on term frequencies." ) elif model_name == "Bi-LSTM": reasoning_parts.append( f"Bidirectional LSTM processed the sequence and identified {phrases_str} as influential." ) if has_negation: reasoning_parts.append( "Sequence context helped partially capture negation through hidden state propagation." ) if word_count > 100: reasoning_parts.append( "Long review — LSTM's recurrent memory tracked sentiment shifts across the sequence." ) elif model_name == "BERT": reasoning_parts.append( f"BERT's self-attention attended to {phrases_str} within full bidirectional context." ) if has_negation: reasoning_parts.append( "Negation was effectively captured via contextual token interactions in attention layers." ) if has_sarcasm: reasoning_parts.append( "Subtle sarcasm/irony signals were detected through contextual embeddings." ) if has_mixed: reasoning_parts.append( "Mixed sentiment detected — BERT resolved the dominant sentiment via attention weighting." ) reasoning_parts.append( f"12 attention heads processed the full sequence context, yielding {conf_label} confidence." ) # Prediction summary sentiment_verb = "positive" if prediction == "Positive" else "negative" reasoning_parts.append( f"Overall: {model_name} classified this review as {sentiment_verb} with {confidence:.1%} confidence." ) return " ".join(reasoning_parts)