src/summarizer.py

import torch
from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
import re
from typing import Optional

class TaskSummarizer:
    def __init__(self, model_name="cointegrated/rut5-base-absum"):
        """
        Инициализация модели для суммаризации
        """
        self.model_name = model_name
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.tokenizer = None
        self.model = None
        
        print(f"🔄 Загрузка модели {model_name}...")
        print(f"📱 Устройство: {self.device}")
        
        try:
            # Загружаем токенизатор и модель
            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
            self.model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
            self.model.to(self.device)
            self.model.eval()  # Режим оценки (не обучения)
            print("✅ Модель успешно загружена!")
        except Exception as e:
            print(f"❌ Ошибка при загрузке модели: {e}")
            print("💡 Попробуйте выполнить: pip install --upgrade transformers torch")
            raise
    
    def summarize(self, text: str, max_length: int = 50, min_length: int = 10) -> str:
        """
        Создает краткую суммаризацию текста задачи
        
        Args:
            text: Полный текст задачи
            max_length: Максимальная длина суммаризации
            min_length: Минимальная длина суммаризации
            
        Returns:
            Краткое описание задачи
        """
        if not text or len(text) < 20:
            return text
        
        try:
            # Очищаем текст от лишних символов
            text = self._clean_text(text)
            
            # Токенизируем входной текст
            inputs = self.tokenizer(
                text, 
                max_length=512, 
                truncation=True, 
                return_tensors="pt"
            ).to(self.device)
            
            # Генерируем суммаризацию
            with torch.no_grad():  # Отключаем вычисление градиентов для экономии памяти
                summary_ids = self.model.generate(
                    inputs.input_ids,
                    max_length=max_length,
                    min_length=min_length,
                    num_beams=4,  # Поиск с лучом для лучшего качества
                    length_penalty=2.0,  # Штраф за длину
                    early_stopping=True,
                    no_repeat_ngram_size=3  # Избегаем повторений
                )
            
            # Декодируем результат
            summary = self.tokenizer.decode(summary_ids[0], skip_special_tokens=True)
            
            # Постобработка
            summary = self._postprocess_summary(summary)
            
            return summary
            
        except Exception as e:
            print(f"⚠️ Ошибка при суммаризации: {e}")
            # Возвращаем первые 100 символов как запасной вариант
            return text[:100] + "..."
    
    def _clean_text(self, text: str) -> str:
        """Очищает текст от лишних символов"""
        # Удаляем номер задачи в начале (если есть)
        text = re.sub(r'^\d+\.\s*', '', text)
        
        # Удаляем информацию об ответственном и сроке
        text = re.sub(r'Отв\.:.*?Срок\s*-\s*\d{2}\.\d{2}\.\d{4}', '', text)
        text = re.sub(r'Отв\.:.*$', '', text, flags=re.MULTILINE)
        text = re.sub(r'Срок\s*-\s*\d{2}\.\d{2}\.\d{4}', '', text)
        
        # Удаляем лишние пробелы
        text = re.sub(r'\s+', ' ', text)
        
        return text.strip()
    
    def _postprocess_summary(self, summary: str) -> str:
        """Постобработка сгенерированной суммаризации"""
        # Убираем лишние пробелы
        summary = re.sub(r'\s+', ' ', summary)
        
        # Убираем точку в конце, если её нет
        if summary and not summary.endswith(('.', '!', '?')):
            summary += '.'
        
        # Делаем первую букву заглавной
        if summary:
            summary = summary[0].upper() + summary[1:]
        
        return summary
    
    def summarize_batch(self, texts, max_length=50, min_length=10):
        """
        Суммаризация нескольких текстов (для эффективности)
        """
        summaries = []
        for text in texts:
            summary = self.summarize(text, max_length, min_length)
            summaries.append(summary)
        return summaries