app.py

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("""
    <style>
    .feature-badge {
        display: inline-block;
        padding: 0.25em 0.6em;
        font-size: 75%;
        font-weight: 700;
        line-height: 1;
        text-align: center;
        white-space: nowrap;
        vertical-align: baseline;
        border-radius: 0.25rem;
        margin-right: 5px;
        margin-bottom: 5px;
    }
    .positive-badge { background-color: #4CAF50; color: #00000;}
    .negative-badge { background-color: #F44336; color: #00000;}
    .neutral-badge { background-color: #2196F3; color: #00000;}
    .feature-badge-default { background-color: #6c757d; color: #00000;}
    .header-box {
        border-radius: 10px;
        padding: 20px;
        margin-bottom: 20px;
        box-shadow: 0 4px 6px rgba(0,0,0,0.1);
    }
    .success-box { border-left: 5px solid #4CAF50; }
    .info-box { border-left: 5px solid #2196F3; }
    .warning-box { border-left: 5px solid #ffc107; }
    .danger-box { #ffebee; border-left: 5px solid #F44336; }
    </style>
""", 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"""
            <div class="header-box {sentiment_class}">
                <h2 style="margin:0;">{report['sentiment']}</h2>
                <p style="font-size: 1.2rem; margin:0;">Confidence: <b>{report['confidence']}</b></p>
            </div>
            """, 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"<div class='feature-badge feature-badge-default'>{feature}</div>", 
                                   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")