import streamlit as st import pandas as pd import tensorflow as tf import numpy as np import pickle import os import re import emoji import contractions import nltk from nltk.tokenize import word_tokenize from nltk.corpus import stopwords import time import matplotlib.pyplot as plt from wordcloud import WordCloud from collections import Counter import tensorflow.keras.backend as K # Download NLTK resources nltk.download('punkt', quiet=True) nltk.download('stopwords', quiet=True) # --- Custom Layers --- @tf.keras.utils.register_keras_serializable(package="CustomLayers") class FeatureExtractor(tf.keras.layers.Layer): def __init__(self, **kwargs): super(FeatureExtractor, self).__init__(**kwargs) def build(self, input_shape): # We'll create trainable weights for feature detection self.contrast_kernel = self.add_weight(name='contrast_kernel', shape=(input_shape[-1], 1), initializer='glorot_uniform') self.negation_kernel = self.add_weight(name='negation_kernel', shape=(input_shape[-1], 1), initializer='glorot_uniform') self.intensifier_kernel = self.add_weight(name='intensifier_kernel', shape=(input_shape[-1], 1), initializer='glorot_uniform') super(FeatureExtractor, self).build(input_shape) def call(self, inputs): # Detect contrast indicators contrast = tf.tensordot(inputs, self.contrast_kernel, axes=1) contrast = tf.squeeze(contrast, axis=-1) contrast = tf.sigmoid(contrast) # Detect negation patterns negation = tf.tensordot(inputs, self.negation_kernel, axes=1) negation = tf.squeeze(negation, axis=-1) negation = tf.sigmoid(negation) # Detect intensifiers/diminishers intensifier = tf.tensordot(inputs, self.intensifier_kernel, axes=1) intensifier = tf.squeeze(intensifier, axis=-1) intensifier = tf.sigmoid(intensifier) # Combine features features = tf.stack([contrast, negation, intensifier], axis=-1) return features def compute_output_shape(self, input_shape): return (input_shape[0], input_shape[1], 3) # (batch_size, seq_length, 3 features) @tf.keras.utils.register_keras_serializable(package="CustomLayers") class SentimentAdjuster(tf.keras.layers.Layer): def __init__(self, **kwargs): super(SentimentAdjuster, self).__init__(**kwargs) def build(self, input_shape): self.contrast_weight = self.add_weight( name='contrast_weight', shape=(3,), initializer='zeros' ) self.negation_weight = self.add_weight( name='negation_weight', shape=(3,), initializer='zeros' ) super(SentimentAdjuster, self).build(input_shape) def call(self, inputs): predictions, features = inputs # Aggregate features (max pooling) contrast_features = tf.reduce_max(features[..., 0], axis=1) negation_features = tf.reduce_max(features[..., 1], axis=1) intensifier_features = tf.reduce_max(features[..., 2], axis=1) # Rule 1: Contrast adjustment contrast_mask = tf.cast(contrast_features > 0.5, tf.float32) contrast_adjustment = contrast_mask * self.contrast_weight[0] # Rule 2: Negation adjustment negation_mask = tf.cast(negation_features > 0.5, tf.float32) negation_adjustment = negation_mask * self.negation_weight[0] # Rule 3: Intensifier adjustment intensifier_mask = tf.cast(intensifier_features > 0.5, tf.float32) intensifier_adjustment = intensifier_mask * self.contrast_weight[1] # Combine adjustments total_adjustment = contrast_adjustment + negation_adjustment + intensifier_adjustment # Create adjustment matrix adjustment_matrix = tf.stack([ total_adjustment * self.contrast_weight[2], # Positive adjustment tf.zeros_like(total_adjustment), # Neutral adjustment -total_adjustment * self.negation_weight[1] # Negative adjustment ], axis=1) # Apply adjustments adjusted = predictions + adjustment_matrix # Ensure valid probabilities adjusted = tf.clip_by_value(adjusted, 1e-7, 1 - 1e-7) adjusted = adjusted / tf.reduce_sum(adjusted, axis=1, keepdims=True) return adjusted def compute_output_shape(self, input_shape): # Same as predictions shape return input_shape[0] @tf.keras.utils.register_keras_serializable(package="CustomLayers") class SimpleAttention(tf.keras.layers.Layer): def __init__(self, **kwargs): super(SimpleAttention, self).__init__(**kwargs) def build(self, input_shape): self.W = self.add_weight( name="attention_weight", shape=(input_shape[-1], 1), initializer="glorot_uniform", trainable=True ) super(SimpleAttention, self).build(input_shape) def call(self, inputs): e = K.tanh(K.dot(inputs, self.W)) e = K.squeeze(e, axis=-1) alpha = K.softmax(e, axis=1) alpha = K.expand_dims(alpha, axis=-1) context = inputs * alpha return K.sum(context, axis=1) def compute_output_shape(self, input_shape): return (input_shape[0], input_shape[2]) # --- Text Preprocessing --- def preprocess_for_lstm(text, remove_stopwords=False): if not isinstance(text, str) or not text.strip(): return "" try: # Handle neutral/negation phrases neutral_phrases = [ 'not bad', 'not great', 'okay', 'so-so', 'meh', 'average', 'mediocre', 'acceptable', 'tolerable', 'passable', 'decent', 'nothing special', 'middle of the road', 'run of the mill' ] for phrase in neutral_phrases: text = re.sub(r'\b' + re.escape(phrase) + r'\b', ' neutral_term ', text, flags=re.IGNORECASE) # Enhanced negation handling negation_patterns = [ r'\b(not|no|never|without|nobody|none|nothing|nowhere|neither|nor)\b [\w]+', r'\b(less than|barely|hardly|scarcely|rarely|seldom)\b [\w]+', r'\b(avoid|skip|doubt|problem|issue|complaint|warning|caution|refuse)\b', r'\b(despite|in spite of|regardless|although|even though)\b' ] for pattern in negation_patterns: text = re.sub(pattern, ' negation_term ', text, flags=re.IGNORECASE) # Emoji handling text = emoji.demojize(text, delimiters=("", "")) # Contractions text = contractions.fix(text) # URL/mention replacement text = re.sub(r'https?://\S+|www\.\S+', ' URL ', text) text = re.sub(r'@\S+', ' USER ', text) text = re.sub(r'\s+', ' ', text).strip().lower() # Emoticon preservation emoticons = re.findall(r'(?::|;|=)(?:-)?(?:\)|\(|D|P)', text) text = re.sub(r'[^\w\s!?.,]', ' ', text) # Tokenization with advanced handling tokens = word_tokenize(text) processed_tokens = [] # Contextual sentiment indicators contextual_indicators = { 'but': 'contrast_indicator', 'however': 'contrast_indicator', 'although': 'contrast_indicator', 'except': 'contrast_indicator', 'unless': 'contrast_indicator', 'yet': 'contrast_indicator', 'still': 'contrast_indicator', 'nonetheless': 'contrast_indicator', 'very': 'intensifier', 'extremely': 'intensifier', 'absolutely': 'intensifier', 'completely': 'intensifier', 'utterly': 'intensifier', 'slightly': 'diminisher', 'somewhat': 'diminisher', 'barely': 'diminisher', 'marginally': 'diminisher', 'almost': 'diminisher', 'only': 'diminisher', 'wow': 'positive_exclamation', 'awesome': 'positive_exclamation', 'ugh': 'negative_exclamation', 'yuck': 'negative_exclamation' } for token in tokens: if not token.strip(): continue # Handle contextual indicators if token in contextual_indicators: processed_tokens.append(contextual_indicators[token]) continue processed_tokens.append(token) processed_tokens.extend(emoticons) return ' '.join(processed_tokens) except Exception: return text.lower() # --- Load model resources --- @st.cache_resource def load_model(): MODEL_DIR = "model_files/models" model_path = f"{MODEL_DIR}/enhanced_lstm_20250624-222759_best.keras" tokenizer_path = f"{MODEL_DIR}/enhanced_lstm_20250624-222759_tokenizer.pickle" label_mapping_path = f"{MODEL_DIR}/enhanced_lstm_20250624-222759_label_mapping.pickle" # Verify files exist for path in [model_path, tokenizer_path, label_mapping_path]: if not os.path.exists(path): st.error(f"Critical error: File not found - {path}") st.stop() # Load model with custom layers try: model = tf.keras.models.load_model( model_path, custom_objects={ 'SimpleAttention': SimpleAttention, 'FeatureExtractor': FeatureExtractor, 'SentimentAdjuster': SentimentAdjuster }, compile=False ) except Exception as e: st.error(f"Model loading failed: {str(e)}") st.stop() # Load tokenizer and label mapping with open(tokenizer_path, "rb") as handle: tokenizer = pickle.load(handle) with open(label_mapping_path, "rb") as handle: label_mapping = pickle.load(handle) return model, tokenizer, label_mapping # --- Initialize resources --- try: MAX_LEN = 50 model, tokenizer, label_mapping = load_model() # Sentiment label mapping SENTIMENT_MAP = { "1.0": {"display": "Positive ๐Ÿ˜Š", "color": "#4CAF50", "name": "Positive"}, "0.0": {"display": "Neutral ๐Ÿ˜", "color": "#2196F3", "name": "Neutral"}, "-1.0": {"display": "Negative ๐Ÿ˜ ", "color": "#F44336", "name": "Negative"} } # Reverse mapping for labels index_to_label = {v: k for k, v in label_mapping.items()} except Exception as e: st.error(f"Initialization failed: {str(e)}") st.stop() # --- Prediction Pipeline --- def predict_sentiment(text): start_time = time.time() processed_text = preprocess_for_lstm(text) # Handle empty sequences if not processed_text.strip(): return "0.0", 0.0, processed_text, {} # Tokenize with fallback seq = tokenizer.texts_to_sequences([processed_text]) if not seq or not any(seq[0]): seq = [[tokenizer.word_index.get(tokenizer.oov_token, 1)]] padded = tf.keras.preprocessing.sequence.pad_sequences( seq, maxlen=MAX_LEN, padding='post', truncating='post', value=0 ) # Predict try: prediction = model.predict(padded, verbose=0)[0] label_idx = np.argmax(prediction) confidence = np.max(prediction) final_label = index_to_label[label_idx] proc_time = time.time() - start_time # Store debug info debug_info = { "raw_text": text, "processed_text": processed_text, "probabilities": { "Negative": float(prediction[0]), "Neutral": float(prediction[1]), "Positive": float(prediction[2]) }, "predicted_label": final_label, "confidence": float(confidence), "processing_time": proc_time } return final_label, confidence, processed_text, debug_info except Exception as e: return "0.0", 0.0, "", {"error": str(e)} # --- Generate Sentiment Report --- def generate_sentiment_report(label, confidence, debug_info): report = { "sentiment": SENTIMENT_MAP[label]["display"], "confidence": f"{confidence:.1%}", "color": SENTIMENT_MAP[label]["color"], "features": [], "key_phrases": [], "word_cloud": None } if not debug_info: return report # Feature explanations feature_explanations = { "has_positive_booster": "Positive language boosters detected", "has_negative_amplifier": "Negative sentiment amplifiers present", "has_neutral_term": "Neutral terms identified", "has_negation_term": "Negation patterns found", "has_contrast_indicator": "Contrast indicators present", "has_intensifier": "Intensifying words used", "has_diminisher": "Diminishing words used", "has_positive_exclamation": "Positive exclamations detected", "has_negative_exclamation": "Negative exclamations found" } for feature, explanation in feature_explanations.items(): if debug_info.get("features", {}).get(feature, False): report["features"].append(explanation) # Key phrase extraction processed_text = debug_info.get("processed_text", "") special_phrases = [ 'neutral_term', 'negation_term', 'contrast_indicator', 'intensifier', 'diminisher', 'positive_booster', 'negative_amplifier', 'positive_exclamation', 'negative_exclamation' ] for phrase in special_phrases: if phrase in processed_text: report["key_phrases"].append(phrase.replace('_', ' ').title()) # Generate word cloud try: wordcloud = WordCloud( width=400, height=200, background_color='white', colormap='viridis', max_words=30 ).generate(processed_text) plt.figure(figsize=(8, 4), facecolor=None) plt.imshow(wordcloud) plt.axis("off") plt.tight_layout(pad=0) report["word_cloud"] = plt except Exception: report["word_cloud"] = None return report # --- Streamlit App UI --- st.set_page_config( page_title="Sentiment Analyzer", layout="wide", page_icon="๐Ÿ“Š" ) st.title("๐Ÿ“Š Sentiment Analysis") st.markdown(""" """, unsafe_allow_html=True) # Initialize session state if 'last_prediction' not in st.session_state: st.session_state.last_prediction = None if 'analysis_history' not in st.session_state: st.session_state.analysis_history = [] # Model info sidebar with st.sidebar: st.header("Model Information") st.write(f"**Model Name:** Optimized LSTM") st.write(f"**Classes:**") for label, data in SENTIMENT_MAP.items(): st.markdown(f"- {data['display']} `{label}`") st.divider() st.header("Analysis History") if st.session_state.analysis_history: for i, item in enumerate(st.session_state.analysis_history[:5]): st.caption(f"{i+1}. {item['text'][:50]}... โ†’ {SENTIMENT_MAP[item['label']]['display']}") else: st.caption("No history yet") # Validation tests with explanations test_cases = [ ("I love this product! It's absolutely amazing ๐Ÿ˜", "1.0", "Clear positive"), ("Terrible experience, worst purchase ever", "-1.0", "Clear negative"), ("The item is okay, nothing special", "0.0", "Neutral - baseline"), ("Not bad but could be better", "0.0", "Neutral - nuanced"), ("Service was not great", "0.0", "Neutral - negation"), ("Best decision I've ever made!", "1.0", "Positive - intensifier"), ("The product is good but the service is terrible", "0.0", "Mixed sentiment"), ("I'm extremely satisfied with my purchase", "1.0", "Positive with intensifier"), ("Somewhat disappointed with the quality", "-1.0", "slightly negative"), ("Absolutely horrible customer service", "-1.0", "Negative with amplifier"), ("The phone is good, however the battery life is bad", "0.0", "Contrast indicator"), ("Wow! This exceeded all my expectations", "1.0", "Positive exclamation"), ("Ugh, this is disgusting", "-1.0", "Negative exclamation") ] with st.expander("๐Ÿงช Validation Test Section", expanded=True): cols = st.columns([3, 1]) with cols[0]: st.subheader("Temp Validation Tests") with cols[1]: if st.button("Run All Tests", type="primary", key="run_tests"): test_results = [] with st.spinner("Running validation suite..."): for text, expected, desc in test_cases: label, confidence, _, debug_info = predict_sentiment(text) match = label == expected test_results.append({ "Text": text, "Description": desc, "Expected": SENTIMENT_MAP[expected]["display"], "Predicted": SENTIMENT_MAP[label]["display"], "Confidence": f"{confidence:.1%}", "Result": "Pass โœ“" if match else "Fail โœ—" }) # Display results df_results = pd.DataFrame(test_results) # Color coding def color_result(val): color = 'green' if val == "Pass โœ“" else 'red' return f'color: {color}; font-weight: bold' st.dataframe( df_results.style.applymap( lambda x: color_result(x) if x in ["Pass โœ“", "Fail โœ—"] else '' ) ) # Calculate pass rate pass_rate = (df_results["Result"] == "Pass โœ“").mean() st.metric("Validation Score", f"{pass_rate:.1%}", delta=f"{len(test_cases)} tests", delta_color="normal") # Single text analysis with st.form("analysis_form", clear_on_submit=False): st.subheader("๐Ÿ” Text Analysis") user_input = st.text_area("Enter text:", height=150, value="The product quality is excellent") submitted = st.form_submit_button("Analyze Sentiment", type="primary", use_container_width=True) if submitted and user_input.strip(): with st.spinner("Analyzing text..."): label, confidence, processed_text, debug_info = predict_sentiment(user_input) # Save to history st.session_state.analysis_history.insert(0, { "text": user_input, "label": label, "confidence": confidence, "timestamp": time.time() }) # Generate report report = generate_sentiment_report(label, confidence, debug_info) st.session_state.last_prediction = debug_info # Display results sentiment_class = "success-box" if label == "1.0" else \ "danger-box" if label == "-1.0" else "info-box" st.markdown(f"""

{report['sentiment']}

Confidence: {report['confidence']}

""", unsafe_allow_html=True) # Feature badges if report["features"]: st.subheader("Key Features Detected") cols = st.columns(3) for i, feature in enumerate(report["features"]): with cols[i % 3]: st.markdown(f"
{feature}
", unsafe_allow_html=True) # Word cloud and probabilities col1, col2 = st.columns(2) with col1: if report["word_cloud"]: st.subheader("Keyword Analysis") st.pyplot(report["word_cloud"]) with col2: st.subheader("Sentiment Probabilities") if st.session_state.last_prediction and "probabilities" in st.session_state.last_prediction: prob_data = { "Negative": st.session_state.last_prediction['probabilities']["Negative"], "Neutral": st.session_state.last_prediction['probabilities']["Neutral"], "Positive": st.session_state.last_prediction['probabilities']["Positive"] } st.bar_chart(prob_data) # Confidence indicator st.metric("Confidence Level", report["confidence"], delta="High confidence" if confidence > 0.8 else "Medium confidence" if confidence > 0.65 else "Low confidence") # Debug info with st.expander("Analysis Details"): st.write(f"**Processed Text:**") st.code(processed_text) if st.session_state.last_prediction: st.write("**Debug Information:**") st.json(st.session_state.last_prediction) # --- CSV Batch Processing Section --- st.subheader("๐Ÿ“Š Batch Analysis from CSV") st.write("Analyze large datasets by uploading a CSV file with text column") uploaded_file = st.file_uploader("Upload CSV file", type=["csv"], help="File must contain a column named 'text'") if uploaded_file is not None: try: # Read CSV file df = pd.read_csv(uploaded_file) # Verify required column exists if 'text' not in df.columns: st.error("โŒ CSV file must contain a column named 'text'") st.stop() st.success(f"โœ… Successfully loaded {len(df)} records") with st.expander("Preview Data", expanded=True): st.dataframe(df.head(3)) # Process in batches if st.button("Analyze Entire Dataset", type="primary", key="batch_analyze"): results = [] sentiment_counts = Counter() feature_counts = Counter() progress_bar = st.progress(0) status_text = st.empty() status_placeholder = st.empty() # Process each row for i, row in enumerate(df.itertuples()): text = str(row.text) label, confidence, _, debug_info = predict_sentiment(text) # Get sentiment name sentiment_name = SENTIMENT_MAP[label]["name"] sentiment_counts[sentiment_name] += 1 # Count features if debug_info and "features" in debug_info: for feature, present in debug_info["features"].items(): if present: feature_counts[feature.replace('_', ' ').title()] += 1 # Add to results results.append({ "Original Text": text, "Processed Text": debug_info.get("processed_text", ""), "Sentiment": sentiment_name, "Label": label, "Confidence": confidence, "Features": ", ".join([ k.replace('_', ' ').title() for k, v in debug_info.get("features", {}).items() if v ]) }) # Update progress progress = (i + 1) / len(df) progress_bar.progress(progress) status_text.text(f"Processed {i+1}/{len(df)} records ({progress:.0%})") # Update every 50 records if i % 50 == 0: with status_placeholder.container(): st.caption(f"Current distribution: {dict(sentiment_counts)}") # Create results dataframe results_df = pd.DataFrame(results) # Show summary st.subheader("Analysis Summary") col1, col2, col3 = st.columns(3) with col1: st.metric("Total Records", len(df)) with col2: st.metric("Positive", f"{sentiment_counts['Positive']} ({sentiment_counts['Positive']/len(df):.1%})") with col3: st.metric("Negative", f"{sentiment_counts['Negative']} ({sentiment_counts['Negative']/len(df):.1%})") # Sentiment distribution st.subheader("Sentiment Distribution") dist_col1, dist_col2 = st.columns([1, 2]) with dist_col1: st.dataframe(pd.DataFrame.from_dict(sentiment_counts, orient='index', columns=['Count'])) with dist_col2: st.bar_chart(pd.Series(sentiment_counts)) # Feature prevalence st.subheader("Feature Frequency") if feature_counts: feature_df = pd.DataFrame.from_dict(feature_counts, orient='index', columns=['Count']) feature_df = feature_df.sort_values('Count', ascending=False) st.dataframe(feature_df) else: st.info("No linguistic features detected in this dataset") # Show results table st.subheader("Detailed Results") st.dataframe(results_df) # Download results csv = results_df.to_csv(index=False).encode('utf-8') st.download_button( label="Download Full Results as CSV", data=csv, file_name="sentiment_analysis_results.csv", mime="text/csv", type="primary" ) except Exception as e: st.error(f"Error processing CSV file: {str(e)}") # Footer st.markdown("---") st.caption("Al-Saadi Sentiment Analysis System v3.4")