app.py

import gradio as gr
import torch
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import time
from datetime import datetime
import pandas as pd
import numpy as np
from typing import List, Dict, Tuple
import logging

# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Кэш для истории запросов
request_history = []
MAX_HISTORY = 10

# Доступные модели для анализа тональности
MODELS = {
    "rubert-tiny-sentiment": {
        "name": "cointegrated/rubert-tiny-sentiment-balanced",
        "description": "Лёгкая модель для русского языка (нейтрал/позитив/негатив)",
        "max_length": 512
    },
    "english-distilbert": {
        "name": "distilbert-base-uncased-finetuned-sst-2-english",
        "description": "Быстрая модель для английского языка (позитив/негатив)",
        "max_length": 512
    },
    "multilingual-xlm": {
        "name": "nlptown/bert-base-multilingual-uncased-sentiment",
        "description": "Мультиязычная модель (1-5 звёзд)",
        "max_length": 512
    }
}

class SentimentAnalyzer:
    def __init__(self):
        self.models = {}
        self.tokenizers = {}
        self.current_model = None
        self.current_model_name = None
        
    def load_model(self, model_key: str):
        """Загрузка модели по ключу"""
        if model_key in self.models and model_key == self.current_model_name:
            return self.models[model_key], self.tokenizers[model_key]
        
        model_info = MODELS[model_key]
        
        try:
            logger.info(f"Загрузка модели: {model_info['name']}")
            
            # Загружаем токенизатор и модель
            tokenizer = AutoTokenizer.from_pretrained(model_info['name'])
            model = AutoModelForSequenceClassification.from_pretrained(model_info['name'])
            
            # Создаем pipeline
            sentiment_pipeline = pipeline(
                "sentiment-analysis",
                model=model,
                tokenizer=tokenizer,
                device=-1  # CPU mode для Spaces
            )
            
            # Сохраняем в кэш
            self.models[model_key] = sentiment_pipeline
            self.tokenizers[model_key] = tokenizer
            self.current_model = sentiment_pipeline
            self.current_model_name = model_key
            
            return sentiment_pipeline, tokenizer
            
        except Exception as e:
            logger.error(f"Ошибка загрузки модели: {e}")
            raise
    
    def analyze_sentiment(self, 
                         text: str, 
                         model_key: str = "rubert-tiny-sentiment",
                         return_raw: bool = False) -> Dict:
        """Анализ тональности текста"""
        
        if not text or not text.strip():
            return {"error": "Текст не может быть пустым"}
        
        # Ограничение длины текста
        if len(text) > 2000:
            text = text[:2000]
            logger.warning(f"Текст обрезан до 2000 символов")
        
        start_time = time.time()
        
        try:
            # Загружаем модель
            pipeline_obj, tokenizer = self.load_model(model_key)
            
            # Выполняем анализ
            result = pipeline_obj(text[:MODELS[model_key]["max_length"]])
            
            processing_time = time.time() - start_time
            
            # Форматируем результат
            if isinstance(result, list):
                result = result[0]
            
            response = {
                "text": text,
                "sentiment": result.get('label', 'N/A'),
                "confidence": round(result.get('score', 0), 4),
                "processing_time": round(processing_time, 3),
                "model": model_key,
                "model_description": MODELS[model_key]["description"]
            }
            
            # Добавляем в историю
            self._add_to_history(response)
            
            return response if return_raw else self._format_output(response)
            
        except Exception as e:
            logger.error(f"Ошибка анализа: {e}")
            return {"error": f"Ошибка обработки: {str(e)}"}
    
    def batch_analyze(self, texts: List[str], model_key: str = "rubert-tiny-sentiment") -> List[Dict]:
        """Пакетный анализ текстов"""
        results = []
        for text in texts:
            if text and text.strip():
                result = self.analyze_sentiment(text, model_key, return_raw=True)
                results.append(result)
        return results
    
    def _format_output(self, result: Dict) -> str:
        """Форматирование вывода"""
        if "error" in result:
            return f"❌ Ошибка: {result['error']}"
        
        # Определяем эмодзи для тональности
        sentiment = result["sentiment"].lower()
        emoji = "😊" if "pos" in sentiment or "позитив" in sentiment or "4" in sentiment or "5" in sentiment else \
                "😐" if "нейтр" in sentiment or "3" in sentiment or "2" in sentiment else \
                "😞" if "негатив" in sentiment or "neg" in sentiment or "1" in sentiment else "🤔"
        
        output = f"""
{emoji} **Результат анализа тональности**

📝 **Текст:** {result['text'][:100]}...

🎭 **Тональность:** {result['sentiment']}
📊 **Уверенность:** {result['confidence'] * 100:.1f}%
⏱️ **Время обработки:** {result['processing_time']} сек.
🤖 **Модель:** {result['model_description']}
        """
        return output
    
    def _add_to_history(self, result: Dict):
        """Добавление запроса в историю"""
        history_entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "text": result["text"][:50] + "..." if len(result["text"]) > 50 else result["text"],
            "sentiment": result.get("sentiment", "N/A"),
            "confidence": result.get("confidence", 0),
            "model": result.get("model", "N/A")
        }
        
        request_history.insert(0, history_entry)
        if len(request_history) > MAX_HISTORY:
            request_history.pop()

# Инициализация анализатора
analyzer = SentimentAnalyzer()

# Примеры для демонстрации
EXAMPLES = [
    ["Это просто потрясающий сервис! Очень доволен качеством обслуживания."],
    ["Ужасный опыт, никогда больше не вернусь. Все работает плохо."],
    ["Сегодня обычный день, ничего особенного не произошло."],
    ["The product is amazing! I love it so much!"],
    ["This is the worst purchase I've ever made."],
    ["El servicio es bastante aceptable, podría ser mejor."]
]

def analyze_text(text: str, model_choice: str, show_history: bool):
    """Основная функция для Gradio"""
    if not text or not text.strip():
        return "⚠️ Пожалуйста, введите текст для анализа."
    
    # Анализ тональности
    result = analyzer.analyze_sentiment(text, model_choice)
    
    # Добавляем историю если нужно
    history_output = ""
    if show_history and request_history:
        history_df = pd.DataFrame(request_history)
        history_output = "\n\n--- 📜 История запросов ---\n"
        history_output += history_df.to_string(index=False)
    
    return result + history_output

def batch_process(file):
    """Обработка пакета текстов из файла"""
    try:
        if file.name.endswith('.txt'):
            with open(file.name, 'r', encoding='utf-8') as f:
                texts = [line.strip() for line in f if line.strip()]
        elif file.name.endswith('.csv'):
            df = pd.read_csv(file.name)
            # Предполагаем, что тексты в первом столбце
            texts = df.iloc[:, 0].dropna().astype(str).tolist()
        else:
            return "❌ Поддерживаются только TXT и CSV файлы"
        
        if not texts:
            return "❌ Файл не содержит текстов для анализа"
        
        # Анализируем все тексты
        results = analyzer.batch_analyze(texts[:20])  # Ограничиваем 20 текстами
        
        # Создаем DataFrame для отображения
        df_results = pd.DataFrame(results)
        
        # Сохраняем результаты
        output_file = "sentiment_results.csv"
        df_results.to_csv(output_file, index=False, encoding='utf-8')
        
        # Статистика
        stats = df_results['sentiment'].value_counts()
        stats_text = "📊 **Статистика результатов:**\n"
        for sentiment, count in stats.items():
            stats_text += f"{sentiment}: {count} текстов\n"
        
        return f"✅ Проанализировано {len(results)} текстов\n\n{stats_text}\n\n📥 Результаты сохранены в: {output_file}"
        
    except Exception as e:
        logger.error(f"Ошибка пакетной обработки: {e}")
        return f"❌ Ошибка обработки файла: {str(e)}"

def calculate_metrics():
    """Расчет метрик качества на тестовых примерах"""
    test_cases = [
        ("Это отличный продукт!", "POSITIVE"),
        ("Ужасное качество", "NEGATIVE"),
        ("Сегодня обычный день", "NEUTRAL"),
        ("I love this movie", "POSITIVE"),
        ("This is terrible", "NEGATIVE")
    ]
    
    correct = 0
    results = []
    
    for text, expected in test_cases:
        result = analyzer.analyze_sentiment(text, "rubert-tiny-sentiment", return_raw=True)
        predicted = result.get("sentiment", "")
        
        # Упрощенная проверка (в реальности нужна нормализация)
        is_correct = expected.lower() in predicted.lower() or predicted.lower() in expected.lower()
        if is_correct:
            correct += 1
        
        results.append({
            "Текст": text[:30] + "...",
            "Ожидалось": expected,
            "Получено": predicted,
            "Верно": "✅" if is_correct else "❌"
        })
    
    accuracy = correct / len(test_cases)
    
    return pd.DataFrame(results), f"📈 Accuracy на тестовых данных: {accuracy:.2%}"

# Создание интерфейса Gradio
with gr.Blocks(title="Анализатор тональности текста", theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 📊 Анализатор тональности текста")
    gr.Markdown("Определите эмоциональную окраску текста: позитив, негатив или нейтрал")
    
    with gr.Row():
        with gr.Column(scale=2):
            text_input = gr.Textbox(
                label="Введите текст для анализа",
                placeholder="Напишите здесь ваш текст...",
                lines=5,
                max_lines=10
            )
            
            model_choice = gr.Dropdown(
                choices=list(MODELS.keys()),
                value="rubert-tiny-sentiment",
                label="Выберите модель",
                info="Выберите модель для анализа тональности"
            )
            
            with gr.Row():
                analyze_btn = gr.Button("🔍 Анализировать", variant="primary")
                clear_btn = gr.Button("Очистить")
            
            show_history = gr.Checkbox(
                label="Показать историю запросов",
                value=False
            )
            
            gr.Markdown("### 📁 Пакетная обработка")
            file_input = gr.File(
                label="Загрузите TXT или CSV файл",
                file_types=[".txt", ".csv"]
            )
            batch_btn = gr.Button("📊 Обработать файл")
            
        with gr.Column(scale=2):
            output_text = gr.Markdown(label="Результат анализа")
            
            gr.Markdown("### 📋 Примеры")
            gr.Examples(
                examples=EXAMPLES,
                inputs=[text_input],
                label="Попробуйте эти примеры:"
            )
            
            gr.Markdown("### 📈 Метрики качества")
            metrics_btn = gr.Button("📏 Рассчитать метрики")
            metrics_output = gr.Markdown()
            metrics_table = gr.Dataframe(label="Результаты тестирования")
    
    # Обработчики событий
    analyze_btn.click(
        fn=analyze_text,
        inputs=[text_input, model_choice, show_history],
        outputs=output_text
    )
    
    batch_btn.click(
        fn=batch_process,
        inputs=file_input,
        outputs=output_text
    )
    
    metrics_btn.click(
        fn=calculate_metrics,
        outputs=[metrics_table, metrics_output]
    )
    
    clear_btn.click(
        fn=lambda: ("", ""),
        outputs=[text_input, output_text]
    )
    
    gr.Markdown("---")
    gr.Markdown("### ℹ️ Информация")
    gr.Markdown("""
    - **Ограничения:** Максимальная длина текста - 2000 символов
    - **Модели:** Поддерживаются русский, английский и другие языки
    - **История:** Сохраняются последние 10 запросов
    - **Форматы:** Поддерживается пакетная обработка TXT и CSV файлов
    """)

if __name__ == "__main__":
    demo.launch(share=False, server_name="0.0.0.0", server_port=7860)