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)