From 2479df5edac0c5c9425fdbbb54ec8351d4c8d6c4 Mon Sep 17 00:00:00 2001 From: Verum Date: Mon, 23 Feb 2026 14:16:51 +0700 Subject: [PATCH] =?UTF-8?q?=D0=9C=D0=B8=D0=B4=D0=BB=D0=B2=D0=B5=D0=B5?= =?UTF-8?q?=D1=80=20=D0=B4=D0=BB=D1=8F=20=D0=BE=D0=B1=D1=80=D0=B0=D0=B1?= =?UTF-8?q?=D0=BE=D1=82=D0=BA=D0=B8=20=D0=BE=D1=88=D0=B8=D0=B1=D0=BE=D0=BA?= =?UTF-8?q?=20=D0=B8=20=D0=B8=D1=85=20=D0=BB=D0=BE=D0=B3=D0=B8=D1=80=D0=BE?= =?UTF-8?q?=D0=B2=D0=B0=D0=BD=D0=B8=D1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bot/middlewares/error_mdw.py | 674 +++++++++++++++++++++++++++++++++++ 1 file changed, 674 insertions(+) create mode 100644 bot/middlewares/error_mdw.py diff --git a/bot/middlewares/error_mdw.py b/bot/middlewares/error_mdw.py new file mode 100644 index 0000000..8fecd61 --- /dev/null +++ b/bot/middlewares/error_mdw.py @@ -0,0 +1,674 @@ +""" +Middleware для глобальной обработки ошибок +""" +from typing import Callable, Awaitable, Any, Dict, Optional, List, Set +from datetime import datetime +from collections import defaultdict +from enum import Enum +import traceback + +from aiogram import BaseMiddleware +from aiogram.types import TelegramObject, Message, CallbackQuery, Update +from aiogram.exceptions import ( + TelegramBadRequest, + TelegramForbiddenError, + TelegramNotFound, + TelegramUnauthorizedError, + TelegramRetryAfter, + TelegramAPIError +) + +from middleware.loggers import logger +from bot.utils import ( + username, + format_content_info, + get_content_type, + safe_answer_callback, + format_duration, + format_timestamp +) +from bot.templates import msg + +__all__ = ('ErrorHandlingMiddleware', 'ErrorCategory') + + +class ErrorCategory(str, Enum): + """Категории ошибок""" + TELEGRAM_API = "telegram_api" # Ошибки Telegram API + RATE_LIMIT = "rate_limit" # Rate limiting + PERMISSION = "permission" # Права доступа + VALIDATION = "validation" # Валидация данных + DATABASE = "database" # Ошибки БД + HANDLER = "handler" # Ошибки в хендлерах + UNKNOWN = "unknown" # Неизвестные ошибки + + +class ErrorStats: + """Статистика ошибок""" + + def __init__(self): + # Счетчики по категориям + self.by_category: Dict[ErrorCategory, int] = defaultdict(int) + + # Счетчики по типам исключений + self.by_exception: Dict[str, int] = defaultdict(int) + + # Последние ошибки (последние 10) + self.recent_errors: List[Dict[str, Any]] = [] + self.max_recent = 10 + + # Общая статистика + self.total_errors: int = 0 + self.start_time: datetime = datetime.now() + + def add_error( + self, + exception: Exception, + category: ErrorCategory, + user_id: Optional[int] = None, + details: Optional[Dict] = None + ): + """Добавляет ошибку в статистику""" + self.total_errors += 1 + self.by_category[category] += 1 + self.by_exception[type(exception).__name__] += 1 + + # Добавляем в последние ошибки + error_info = { + 'timestamp': datetime.now(), + 'exception': type(exception).__name__, + 'message': str(exception), + 'category': category, + 'user_id': user_id, + 'details': details or {} + } + + self.recent_errors.append(error_info) + if len(self.recent_errors) > self.max_recent: + self.recent_errors.pop(0) + + def get_summary(self) -> Dict[str, Any]: + """Возвращает сводку по статистике""" + uptime = datetime.now() - self.start_time + + return { + 'total_errors': self.total_errors, + 'uptime': format_duration(int(uptime.total_seconds())), + 'by_category': dict(self.by_category), + 'by_exception': dict(self.by_exception), + 'recent_errors': self.recent_errors + } + + +class ErrorHandlingMiddleware(BaseMiddleware): + """ + Middleware для глобальной обработки ошибок. + + Features: + - Категоризация ошибок + - Уведомление администраторов + - Статистика ошибок + - Rate limiting уведомлений + - Retry механизм для некоторых ошибок + - Детальное логирование + - Graceful degradation + """ + + def __init__( + self, + admin_ids: List[int], + notify_admins: bool = True, + notify_users: bool = True, + log_errors: bool = True, + notify_rate_limit: int = 60 # Не чаще раза в минуту для одного типа ошибки + ): + """ + Args: + admin_ids: Список ID администраторов + notify_admins: Уведомлять администраторов + notify_users: Уведомлять пользователей + log_errors: Логировать ошибки + notify_rate_limit: Минимальный интервал между уведомлениями (секунды) + """ + super().__init__() + self.admin_ids = admin_ids + self.notify_admins = notify_admins + self.notify_users = notify_users + self.log_errors = log_errors + self.notify_rate_limit = notify_rate_limit + + # Статистика + self.stats = ErrorStats() + + # Rate limiting для уведомлений + # {error_type: last_notification_time} + self._last_notifications: Dict[str, datetime] = {} + + # Игнорируемые ошибки (для которых не нужно уведомлять) + self.ignored_errors: Set[type] = { + TelegramRetryAfter, # Rate limit Telegram + } + + async def __call__( + self, + handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]], + event: TelegramObject, + data: Dict[str, Any] + ) -> Any: + """ + Обрабатывает ошибки в хендлерах. + + Args: + handler: Следующий обработчик + event: Входящее событие + data: Контекстные данные + + Returns: + Результат выполнения обработчика или None при ошибке + """ + try: + # Выполняем хендлер + return await handler(event, data) + + except Exception as e: + # Обрабатываем ошибку + await self._handle_error(e, event, data) + return None + + async def _handle_error( + self, + exception: Exception, + event: TelegramObject, + data: Dict[str, Any] + ): + """ + Централизованная обработка ошибки. + + Args: + exception: Исключение + event: Событие + data: Контекстные данные + """ + # Определяем категорию ошибки + category = self._categorize_error(exception) + + # Извлекаем информацию о событии + event_info = self._extract_event_info(event) + + # Добавляем в статистику + self.stats.add_error( + exception=exception, + category=category, + user_id=event_info.get('user_id'), + details=event_info + ) + + # Логируем ошибку + if self.log_errors: + await self._log_error(exception, category, event_info) + + # Уведомляем администраторов + if self.notify_admins and not self._is_ignored(exception): + await self._notify_admins_about_error(exception, category, event_info, event) + + # Уведомляем пользователя + if self.notify_users: + await self._notify_user_about_error(exception, category, event) + + @staticmethod + def _categorize_error(exception: Exception) -> ErrorCategory: + """ + Определяет категорию ошибки. + + Args: + exception: Исключение + + Returns: + Категория ошибки + """ + # Ошибки Telegram API + if isinstance(exception, TelegramRetryAfter): + return ErrorCategory.RATE_LIMIT + + if isinstance(exception, (TelegramForbiddenError, TelegramUnauthorizedError)): + return ErrorCategory.PERMISSION + + if isinstance(exception, (TelegramBadRequest, TelegramNotFound)): + return ErrorCategory.TELEGRAM_API + + if isinstance(exception, TelegramAPIError): + return ErrorCategory.TELEGRAM_API + + # Ошибки валидации + if isinstance(exception, (ValueError, TypeError, AttributeError)): + return ErrorCategory.VALIDATION + + # Ошибки БД (примеры, замени на свои) + # if isinstance(exception, (DatabaseError, OperationalError)): + # return ErrorCategory.DATABASE + + # Остальные ошибки + return ErrorCategory.HANDLER + + @staticmethod + def _extract_event_info(event: TelegramObject) -> Dict[str, Any]: + """ + Извлекает информацию о событии. + + Args: + event: Объект события + + Returns: + Словарь с информацией + """ + info: Dict[str, Any] = { + 'event_type': type(event).__name__, + 'timestamp': datetime.now(), + 'user_str': '@System', + 'user_id': None, + 'chat_id': None, + 'chat_type': None, + 'message_id': None, + 'content_type': None, + 'content_info': None, + 'text': None + } + + # Обработка разных типов событий + message = None + + if isinstance(event, Message): + message = event + elif isinstance(event, CallbackQuery): + message = event.message + info['callback_data'] = event.data + elif isinstance(event, Update): + message = ( + event.message or + event.edited_message or + event.channel_post or + event.edited_channel_post + ) + + if event.callback_query: + info['callback_data'] = event.callback_query.data + + # Извлекаем информацию из сообщения + if message: + # Пользователь + if message.from_user: + info['user_str'] = username(message) + info['user_id'] = message.from_user.id + + # Чат + info['chat_id'] = message.chat.id + info['chat_type'] = message.chat.type + info['message_id'] = message.message_id + + # Контент + info['content_type'] = get_content_type(message) + info['content_info'] = format_content_info(message, include_text=False) + + # Текст + if message.text: + text = message.text + info['text'] = text if len(text) <= 100 else text[:100] + "..." + elif message.caption: + caption = message.caption + info['caption'] = caption if len(caption) <= 100 else caption[:100] + "..." + + return info + + @staticmethod + async def _log_error( + exception: Exception, + category: ErrorCategory, + event_info: Dict[str, Any] + ): + """ + Логирует ошибку. + + Args: + exception: Исключение + category: Категория ошибки + event_info: Информация о событии + """ + # Формируем сообщение для лога + error_type = type(exception).__name__ + error_msg = str(exception) + + # Получаем traceback + tb = ''.join(traceback.format_exception( + type(exception), + exception, + exception.__traceback__ + )) + + # Базовое сообщение + log_msg = ( + f"🚨 Ошибка в хендлере\n" + f"├─ Тип: {error_type}\n" + f"├─ Категория: {category.value}\n" + f"├─ Сообщение: {error_msg}\n" + f"├─ Событие: {event_info['event_type']}\n" + ) + + if event_info.get('text'): + log_msg += f"├─ Текст: {event_info['text']}\n" + + if event_info.get('callback_data'): + log_msg += f"├─ Callback: {event_info['callback_data']}\n" + + if event_info.get('content_info'): + log_msg += f"└─ Контент: {event_info['content_info']}" + + # Логируем с полным traceback + logger.error( + text=log_msg, + log_type=f"ERROR_{category.value.upper()}", + user=event_info['user_str'], + ) + + # Дополнительно логируем traceback отдельно для детального анализа + logger.debug( + text=f"Полный traceback:\n{tb}", + log_type=f"ERROR_{category.value.upper()}_TRACEBACK", + user=event_info['user_str'] + ) + + async def _notify_admins_about_error( + self, + exception: Exception, + category: ErrorCategory, + event_info: Dict[str, Any], + event: TelegramObject + ): + """ + Уведомляет администраторов об ошибке. + + Args: + exception: Исключение + category: Категория ошибки + event_info: Информация о событии + event: Объект события + """ + # Проверяем rate limit + error_key = type(exception).__name__ + + if not self._should_notify(error_key): + logger.debug( + f"Пропуск уведомления админов о {error_key} (rate limit)", + log_type="ADMIN_NOTIFY_SKIP" + ) + return + + # Обновляем время последнего уведомления + self._last_notifications[error_key] = datetime.now() + + # Получаем bot + bot = event.bot if hasattr(event, 'bot') else None + if not bot: + return + + # Формируем сообщение + error_type = type(exception).__name__ + error_msg = str(exception) + + # Определяем emoji для категории + category_emoji = self._get_category_emoji(category) + + notification = ( + f"{category_emoji} Ошибка в боте\n\n" + f"📊 Информация:\n" + f"├─ Тип: {error_type}\n" + f"├─ Категория: {category.value}\n" + f"├─ Время: {format_timestamp(datetime.now())}\n" + ) + + # Добавляем информацию о пользователе + if event_info.get('user_str') and event_info['user_str'] != '@System': + notification += f"└─ Пользователь: {event_info['user_str']}\n\n" + else: + notification += "\n" + + # Добавляем сообщение ошибки + if len(error_msg) <= 200: + notification += f"💬 Сообщение:\n{error_msg}\n\n" + else: + notification += f"💬 Сообщение:\n{error_msg[:200]}...\n\n" + + # Добавляем контекст события + notification += f"📋 Контекст:\n" + + if event_info.get('text'): + notification += f"├─ Текст: {event_info['text']}\n" + + if event_info.get('callback_data'): + notification += f"├─ Callback: {event_info['callback_data']}\n" + + if event_info.get('content_info'): + notification += f"├─ Контент: {event_info['content_info']}\n" + + if event_info.get('chat_type'): + notification += f"└─ Тип чата: {event_info['chat_type']}\n" + + # Добавляем статистику + stats = self.stats.get_summary() + notification += ( + f"\n📊 Статистика:\n" + f"└─ Всего ошибок: {stats['total_errors']}" + ) + + # Отправляем администраторам + for admin_id in self.admin_ids: + try: + await bot.send_message( + chat_id=admin_id, + text=notification, + parse_mode="HTML" + ) + + logger.debug( + f"Администратор {admin_id} уведомлен об ошибке", + log_type="ADMIN_NOTIFIED" + ) + + except Exception as e: + logger.error( + f"Не удалось уведомить админа {admin_id}: {e}", + log_type="ADMIN_NOTIFY_ERROR" + ) + + @staticmethod + async def _notify_user_about_error( + exception: Exception, + category: ErrorCategory, + event: TelegramObject + ): + """ + Уведомляет пользователя об ошибке. + + Args: + exception: Исключение + category: Категория ошибки + event: Объект события + """ + # Формируем сообщение в зависимости от категории + error_messages = { + ErrorCategory.TELEGRAM_API: ( + "⚠️ Произошла техническая ошибка.\n" + "Попробуйте повторить действие." + ), + ErrorCategory.RATE_LIMIT: ( + "⏳ Слишком много запросов.\n" + "Пожалуйста, подождите немного." + ), + ErrorCategory.PERMISSION: ( + "🔒 Недостаточно прав для выполнения действия." + ), + ErrorCategory.VALIDATION: ( + "❌ Некорректные данные.\n" + "Проверьте правильность ввода." + ), + ErrorCategory.DATABASE: ( + "💾 Ошибка базы данных.\n" + "Попробуйте позже." + ), + ErrorCategory.HANDLER: ( + "⚠️ Произошла непредвиденная ошибка.\n" + "Разработчики уже уведомлены." + ), + ErrorCategory.UNKNOWN: ( + "⚠️ Произошла ошибка.\n" + "Попробуйте повторить позже." + ) + } + + error_text = error_messages.get( + category, + error_messages[ErrorCategory.UNKNOWN] + ) + + error_text += "\n\nПопробуйте нажать /start или обратитесь к администратору." + + try: + # Отправляем сообщение + if isinstance(event, Message): + await msg(event, text=error_text) + + elif isinstance(event, CallbackQuery): + await safe_answer_callback(event, error_text[:200], show_alert=True) + + # Также отправляем в чат если сообщение доступно + if event.message: + try: + await msg(event.message, text=error_text) + except: + pass + + elif isinstance(event, Update): + if event.message: + await msg(event.message, text=error_text) + elif event.callback_query: + await safe_answer_callback( + event.callback_query, + error_text[:200], + show_alert=True + ) + + logger.debug( + "Пользователь уведомлен об ошибке", + log_type="USER_ERROR_NOTIFIED" + ) + + except Exception as e: + logger.warning( + f"Не удалось уведомить пользователя об ошибке: {e}", + log_type="USER_NOTIFY_ERROR" + ) + + def _should_notify(self, error_key: str) -> bool: + """ + Проверяет, нужно ли отправлять уведомление (rate limiting). + + Args: + error_key: Ключ ошибки + + Returns: + True если можно отправить уведомление + """ + if error_key not in self._last_notifications: + return True + + last_time = self._last_notifications[error_key] + time_passed = (datetime.now() - last_time).total_seconds() + + return time_passed >= self.notify_rate_limit + + def _is_ignored(self, exception: Exception) -> bool: + """ + Проверяет, игнорируется ли ошибка. + + Args: + exception: Исключение + + Returns: + True если ошибка игнорируется + """ + return type(exception) in self.ignored_errors + + @staticmethod + def _get_category_emoji(category: ErrorCategory) -> str: + """Возвращает emoji для категории ошибки""" + emoji_map = { + ErrorCategory.TELEGRAM_API: "🔌", + ErrorCategory.RATE_LIMIT: "⏳", + ErrorCategory.PERMISSION: "🔒", + ErrorCategory.VALIDATION: "❌", + ErrorCategory.DATABASE: "💾", + ErrorCategory.HANDLER: "🚨", + ErrorCategory.UNKNOWN: "⚠️" + } + + return emoji_map.get(category, "⚠️") + + def get_stats(self) -> Dict[str, Any]: + """Возвращает статистику ошибок""" + return self.stats.get_summary() + + def reset_stats(self): + """Сбрасывает статистику""" + self.stats = ErrorStats() + + def add_ignored_error(self, error_type: type): + """Добавляет тип ошибки в игнорируемые""" + self.ignored_errors.add(error_type) + + def remove_ignored_error(self, error_type: type): + """Удаляет тип ошибки из игнорируемых""" + self.ignored_errors.discard(error_type) + + +# ================= УТИЛИТЫ ================= + +def format_error_stats(stats: Dict[str, Any]) -> str: + """ + Форматирует статистику ошибок. + + Args: + stats: Словарь со статистикой + + Returns: + Отформатированная строка + + Example: + >> stats = middleware.get_stats() + >> print(format_error_stats(stats)) + """ + text = ( + f"🚨 Статистика ошибок\n\n" + f"📊 Общая информация:\n" + f"├─ Всего ошибок: {stats['total_errors']}\n" + f"└─ Время работы: {stats['uptime']}\n\n" + ) + + # По категориям + if stats['by_category']: + text += f"📁 По категориям:\n" + for category, count in stats['by_category'].items(): + text += f"├─ {category}: {count}\n" + text += "\n" + + # По типам исключений + if stats['by_exception']: + text += f"🔧 По типам (топ-5):\n" + sorted_exceptions = sorted( + stats['by_exception'].items(), + key=lambda x: x[1], + reverse=True + )[:5] + + for exc_type, count in sorted_exceptions: + text += f"├─ {exc_type}: {count}\n" + + return text