Мидлвеер для обработки ошибок и их логирования
This commit is contained in:
674
bot/middlewares/error_mdw.py
Normal file
674
bot/middlewares/error_mdw.py
Normal file
@@ -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} <b>Ошибка в боте</b>\n\n"
|
||||
f"📊 <b>Информация:</b>\n"
|
||||
f"├─ Тип: <code>{error_type}</code>\n"
|
||||
f"├─ Категория: <code>{category.value}</code>\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"💬 <b>Сообщение:</b>\n<code>{error_msg}</code>\n\n"
|
||||
else:
|
||||
notification += f"💬 <b>Сообщение:</b>\n<code>{error_msg[:200]}...</code>\n\n"
|
||||
|
||||
# Добавляем контекст события
|
||||
notification += f"📋 <b>Контекст:</b>\n"
|
||||
|
||||
if event_info.get('text'):
|
||||
notification += f"├─ Текст: <code>{event_info['text']}</code>\n"
|
||||
|
||||
if event_info.get('callback_data'):
|
||||
notification += f"├─ Callback: <code>{event_info['callback_data']}</code>\n"
|
||||
|
||||
if event_info.get('content_info'):
|
||||
notification += f"├─ Контент: {event_info['content_info']}\n"
|
||||
|
||||
if event_info.get('chat_type'):
|
||||
notification += f"└─ Тип чата: <code>{event_info['chat_type']}</code>\n"
|
||||
|
||||
# Добавляем статистику
|
||||
stats = self.stats.get_summary()
|
||||
notification += (
|
||||
f"\n📊 <b>Статистика:</b>\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"🚨 <b>Статистика ошибок</b>\n\n"
|
||||
f"📊 <b>Общая информация:</b>\n"
|
||||
f"├─ Всего ошибок: {stats['total_errors']}\n"
|
||||
f"└─ Время работы: {stats['uptime']}\n\n"
|
||||
)
|
||||
|
||||
# По категориям
|
||||
if stats['by_category']:
|
||||
text += f"📁 <b>По категориям:</b>\n"
|
||||
for category, count in stats['by_category'].items():
|
||||
text += f"├─ {category}: {count}\n"
|
||||
text += "\n"
|
||||
|
||||
# По типам исключений
|
||||
if stats['by_exception']:
|
||||
text += f"🔧 <b>По типам (топ-5):</b>\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
|
||||
Reference in New Issue
Block a user