deploy_package/app.py

#!/usr/bin/env python3
"""
Vietnamese Sentiment Analysis - Hugging Face Spaces Gradio App
"""

import gradio as gr
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import time
import numpy as np
from datetime import datetime
import gc
import psutil
import os
import pandas as pd

class SentimentGradioApp:
    def __init__(self, model_name="5CD-AI/Vietnamese-Sentiment-visobert", max_batch_size=10):
        self.model_name = model_name
        self.tokenizer = None
        self.model = None
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.sentiment_labels = ["Negative", "Neutral", "Positive"]
        self.sentiment_colors = {
            "Negative": "#ff4444",
            "Neutral": "#ffaa00",
            "Positive": "#44ff44"
        }
        self.model_loaded = False
        self.max_batch_size = max_batch_size
        self.max_memory_mb = 8192  # Hugging Face Spaces memory limit

    def get_memory_usage(self):
        """Get current memory usage in MB"""
        process = psutil.Process(os.getpid())
        return process.memory_info().rss / 1024 / 1024

    def check_memory_limit(self):
        """Check if memory usage is within limits"""
        current_memory = self.get_memory_usage()
        if current_memory > self.max_memory_mb:
            return False, f"Memory usage ({current_memory:.1f}MB) exceeds limit ({self.max_memory_mb}MB)"
        return True, f"Memory usage: {current_memory:.1f}MB"

    def cleanup_memory(self):
        """Clean up GPU and CPU memory"""
        if torch.cuda.is_available():
            torch.cuda.empty_cache()
        gc.collect()

    def load_model(self):
        """Load the model from Hugging Face Hub"""
        if self.model_loaded:
            return True

        try:
            # Clean up any existing memory
            self.cleanup_memory()

            # Check memory before loading
            memory_ok, memory_msg = self.check_memory_limit()
            if not memory_ok:
                print(f"❌ {memory_msg}")
                return False

            print(f"📊 {memory_msg}")
            print(f"🤖 Loading model from Hugging Face Hub: {self.model_name}")

            self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
            self.model = AutoModelForSequenceClassification.from_pretrained(self.model_name)

            self.model.to(self.device)
            self.model.eval()
            self.model_loaded = True

            # Check memory after loading
            memory_ok, memory_msg = self.check_memory_limit()
            print(f"✅ Model loaded successfully from {self.model_name}")
            print(f"📊 {memory_msg}")

            return True
        except Exception as e:
            print(f"❌ Error loading model: {e}")
            self.model_loaded = False
            self.cleanup_memory()
            return False

    def predict_sentiment(self, text):
        """Predict sentiment for given text"""
        if not self.model_loaded:
            return None, "❌ Model not loaded. Please refresh the page."

        if not text.strip():
            return None, "❌ Please enter some text to analyze."

        try:
            # Check memory before prediction
            memory_ok, memory_msg = self.check_memory_limit()
            if not memory_ok:
                return None, f"❌ {memory_msg}"

            start_time = time.time()

            # Tokenize
            inputs = self.tokenizer(
                text,
                return_tensors="pt",
                truncation=True,
                padding=True,
                max_length=512
            )

            # Move to device
            inputs = {k: v.to(self.device) for k, v in inputs.items()}

            # Predict
            with torch.no_grad():
                outputs = self.model(**inputs)
                logits = outputs.logits
                probabilities = torch.softmax(logits, dim=-1)
                predicted_class = torch.argmax(probabilities, dim=-1).item()
                confidence = torch.max(probabilities).item()

            inference_time = time.time() - start_time

            # Move to CPU and clean GPU memory
            probs = probabilities.cpu().numpy()[0].tolist()
            del probabilities, logits, outputs
            self.cleanup_memory()

            sentiment = self.sentiment_labels[predicted_class]

            # Create detailed results
            result = {
                "sentiment": sentiment,
                "confidence": confidence,
                "probabilities": {
                    "Negative": probs[0],
                    "Neutral": probs[1],
                    "Positive": probs[2]
                },
                "inference_time": inference_time,
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            }

            # Create formatted output
            output_text = f"""
## 🎯 Sentiment Analysis Result

**Sentiment:** {sentiment}
**Confidence:** {confidence:.2%}
**Processing Time:** {inference_time:.3f}s

### 📊 Probability Distribution:
- 😠 **Negative:** {probs[0]:.2%}
- 😐 **Neutral:** {probs[1]:.2%}
- 😊 **Positive:** {probs[2]:.2%}

### 📝 Input Text:
> "{text}"

---
*Analysis completed at {result['timestamp']}*
*{memory_msg}*
            """.strip()

            return result, output_text

        except Exception as e:
            self.cleanup_memory()
            return None, f"❌ Error during prediction: {str(e)}"

    def batch_predict(self, texts):
        """Predict sentiment for multiple texts with memory management"""
        if not self.model_loaded:
            return [], "❌ Model not loaded. Please refresh the page."

        if not texts or not any(texts):
            return [], "❌ Please enter some texts to analyze."

        # Filter valid texts and apply batch size limit
        valid_texts = [text.strip() for text in texts if text.strip()]

        if len(valid_texts) > self.max_batch_size:
            return [], f"❌ Too many texts ({len(valid_texts)}). Maximum batch size is {self.max_batch_size} for memory efficiency."

        if not valid_texts:
            return [], "❌ No valid texts provided."

        # Check memory before batch processing
        memory_ok, memory_msg = self.check_memory_limit()
        if not memory_ok:
            return [], f"❌ {memory_msg}"

        results = []
        try:
            for i, text in enumerate(valid_texts):
                # Check memory every 5 predictions
                if i % 5 == 0:
                    memory_ok, memory_msg = self.check_memory_limit()
                    if not memory_ok:
                        break

                result, _ = self.predict_sentiment(text)
                if result:
                    results.append(result)

            if not results:
                return [], "❌ No valid predictions made."

            # Create batch summary
            total_texts = len(results)
            sentiments = [r["sentiment"] for r in results]
            avg_confidence = sum(r["confidence"] for r in results) / total_texts

            sentiment_counts = {
                "Positive": sentiments.count("Positive"),
                "Neutral": sentiments.count("Neutral"),
                "Negative": sentiments.count("Negative")
            }

            summary = f"""
## 📊 Batch Analysis Summary

**Total Texts Analyzed:** {total_texts}/{len(valid_texts)}
**Average Confidence:** {avg_confidence:.2%}
**Memory Used:** {self.get_memory_usage():.1f}MB

### 🎯 Sentiment Distribution:
- 😊 **Positive:** {sentiment_counts['Positive']} ({sentiment_counts['Positive']/total_texts:.1%})
- 😐 **Neutral:** {sentiment_counts['Neutral']} ({sentiment_counts['Neutral']/total_texts:.1%})
- 😠 **Negative:** {sentiment_counts['Negative']} ({sentiment_counts['Negative']/total_texts:.1%})

### 📋 Individual Results:
            """.strip()

            for i, result in enumerate(results, 1):
                summary += f"\n**{i}.** {result['sentiment']} ({result['confidence']:.1%})"

            # Final memory cleanup
            self.cleanup_memory()

            return results, summary

        except Exception as e:
            self.cleanup_memory()
            return [], f"❌ Error during batch processing: {str(e)}"

def create_interface():
    """Create the Gradio interface for Hugging Face Spaces"""
    app = SentimentGradioApp()

    # Load model
    if not app.load_model():
        print("❌ Failed to load model. Please try again.")
        return None

    # Example texts
    examples = [
        "Giảng viên dạy rất hay và tâm huyết.",
        "Môn học này quá khó và nhàm chán.",
        "Lớp học ổn định, không có gì đặc biệt.",
        "Tôi rất thích cách giảng dạy của thầy cô.",
        "Chương trình học cần cải thiện nhiều."
    ]

    # Custom CSS
    css = """
    .gradio-container {
        max-width: 900px !important;
        margin: auto !important;
    }
    .sentiment-positive {
        color: #44ff44;
        font-weight: bold;
    }
    .sentiment-neutral {
        color: #ffaa00;
        font-weight: bold;
    }
    .sentiment-negative {
        color: #ff4444;
        font-weight: bold;
    }
    """

    # Create interface
    with gr.Blocks(
        title="Vietnamese Sentiment Analysis",
        theme=gr.themes.Soft(),
        css=css
    ) as interface:

        gr.Markdown("# 🎭 Vietnamese Sentiment Analysis")
        gr.Markdown("Enter Vietnamese text to analyze sentiment using a transformer model from Hugging Face.")

        with gr.Tabs():
            # Single Text Analysis Tab
            with gr.Tab("📝 Single Text Analysis"):
                with gr.Row():
                    with gr.Column(scale=3):
                        text_input = gr.Textbox(
                            label="Enter Vietnamese Text",
                            placeholder="Type or paste Vietnamese text here...",
                            lines=3
                        )

                        with gr.Row():
                            analyze_btn = gr.Button("🔍 Analyze Sentiment", variant="primary")
                            clear_btn = gr.Button("🗑️ Clear", variant="secondary")

                    with gr.Column(scale=2):
                        gr.Examples(
                            examples=examples,
                            inputs=[text_input],
                            label="💡 Example Texts"
                        )

                result_output = gr.Markdown(label="Analysis Result", visible=True)
                confidence_plot = gr.BarPlot(
                    title="Confidence Scores",
                    x="sentiment",
                    y="confidence",
                    visible=False
                )

            # Batch Analysis Tab
            with gr.Tab("📊 Batch Analysis"):
                gr.Markdown(f"### 📝 Memory-Efficient Batch Processing")
                gr.Markdown(f"**Maximum batch size:** {app.max_batch_size} texts (for memory efficiency)")
                gr.Markdown(f"**Memory limit:** {app.max_memory_mb}MB")

                batch_input = gr.Textbox(
                    label="Enter Multiple Texts (one per line)",
                    placeholder=f"Enter up to {app.max_batch_size} Vietnamese texts, one per line...",
                    lines=8,
                    max_lines=20
                )

                with gr.Row():
                    batch_analyze_btn = gr.Button("🔍 Analyze All", variant="primary")
                    batch_clear_btn = gr.Button("🗑️ Clear", variant="secondary")
                    memory_cleanup_btn = gr.Button("🧹 Memory Cleanup", variant="secondary")

                batch_result_output = gr.Markdown(label="Batch Analysis Result")
                memory_info = gr.Textbox(
                    label="Memory Usage",
                    value=f"{app.get_memory_usage():.1f}MB used",
                    interactive=False
                )

            # Model Info Tab
            with gr.Tab("ℹ️ Model Information"):
                gr.Markdown(f"""
                ## 🤖 Model Details

                **Model Architecture:** Transformer-based sequence classification
                **Base Model:** {app.model_name}
                **Languages:** Vietnamese (optimized)
                **Labels:** Negative, Neutral, Positive
                **Max Batch Size:** {app.max_batch_size} texts

                ## 📊 Performance Metrics

                - **Processing Speed:** ~100ms per text
                - **Max Sequence Length:** 512 tokens
                - **Memory Limit:** {app.max_memory_mb}MB

                ## 💡 Usage Tips

                - Enter clear, grammatically correct Vietnamese text
                - Longer texts (20-200 words) work best
                - The model handles various Vietnamese dialects
                - Confidence scores indicate prediction certainty

                ## 🛡️ Memory Management

                - **Automatic Cleanup:** Memory is cleaned after each prediction
                - **Batch Limits:** Maximum {app.max_batch_size} texts per batch to prevent overflow
                - **Memory Monitoring:** Real-time memory usage tracking
                - **GPU Optimization:** CUDA cache clearing when available

                ## ⚠️ Performance Notes

                - If you encounter memory errors, try reducing batch size
                - Use the Memory Cleanup button if needed
                - Monitor memory usage in the Batch Analysis tab
                - Model loaded directly from Hugging Face Hub (no local training required)
                """)

        # Event handlers
        def analyze_text(text):
            result, output = app.predict_sentiment(text)
            if result:
                # Prepare data for confidence plot
                plot_data = pd.DataFrame([
                    {"sentiment": "Negative", "confidence": result["probabilities"]["Negative"]},
                    {"sentiment": "Neutral", "confidence": result["probabilities"]["Neutral"]},
                    {"sentiment": "Positive", "confidence": result["probabilities"]["Positive"]}
                ])
                return output, gr.BarPlot(visible=True, value=plot_data)
            else:
                return output, gr.BarPlot(visible=False)

        def clear_inputs():
            return "", "", gr.BarPlot(visible=False)

        def analyze_batch(texts):
            if texts:
                text_list = [line.strip() for line in texts.split('\n') if line.strip()]
                results, summary = app.batch_predict(text_list)
                return summary
            return "❌ Please enter some texts to analyze."

        def clear_batch():
            return ""

        def update_memory_info():
            return f"{app.get_memory_usage():.1f}MB used"

        def manual_memory_cleanup():
            app.cleanup_memory()
            return f"Memory cleaned. Current usage: {app.get_memory_usage():.1f}MB"

        # Connect events
        analyze_btn.click(
            fn=analyze_text,
            inputs=[text_input],
            outputs=[result_output, confidence_plot]
        )

        clear_btn.click(
            fn=clear_inputs,
            outputs=[text_input, result_output, confidence_plot]
        )

        batch_analyze_btn.click(
            fn=analyze_batch,
            inputs=[batch_input],
            outputs=[batch_result_output]
        )

        batch_clear_btn.click(
            fn=clear_batch,
            outputs=[batch_input]
        )

        memory_cleanup_btn.click(
            fn=manual_memory_cleanup,
            outputs=[memory_info]
        )

        # Update memory info periodically
        interface.load(
            fn=update_memory_info,
            outputs=[memory_info]
        )

    return interface

# Create and launch the interface
if __name__ == "__main__":
    print("🚀 Starting Vietnamese Sentiment Analysis for Hugging Face Spaces...")

    interface = create_interface()
    if interface is None:
        print("❌ Failed to create interface. Exiting.")
        exit(1)

    print("✅ Interface created successfully!")
    print("🌐 Launching web interface...")

    # Launch the interface
    interface.launch(
        share=True,
        show_error=True,
        quiet=False
    )