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