From b63410187fce935afaf1e09c3efba44a063566f5 Mon Sep 17 00:00:00 2001 From: Verum Date: Mon, 23 Feb 2026 14:16:00 +0700 Subject: [PATCH] =?UTF-8?q?=D0=9D=D0=B0=D1=81=D1=82=D1=80=D0=BE=D0=B9?= =?UTF-8?q?=D0=BA=D0=B0=20=D0=BA=D0=B0=D1=81=D1=82=D0=BE=D0=BC=D0=BD=D0=BE?= =?UTF-8?q?=D0=B3=D0=BE=20=D0=BB=D0=BE=D0=B3=D0=B3=D0=B5=D1=80=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- middleware/loggers/logs.py | 380 +++++++++++++++++++++++++++++++++++++ 1 file changed, 380 insertions(+) create mode 100644 middleware/loggers/logs.py diff --git a/middleware/loggers/logs.py b/middleware/loggers/logs.py new file mode 100644 index 0000000..d2d0566 --- /dev/null +++ b/middleware/loggers/logs.py @@ -0,0 +1,380 @@ +""" +Кастомный логгер с поддержством декораторов и прямого вызова +""" +from sys import stderr +from pathlib import Path +from functools import wraps +from inspect import iscoroutinefunction +from typing import Any, Callable, Optional, TypeVar, Union, cast, Final +from contextlib import contextmanager + +from loguru import logger as nlogger +from aiogram.types import Message, User, CallbackQuery + +from configs import settings + +# Экспортируемые объекты +__all__ = ('Logger', 'setup_logging', 'logger', 'log') + +# Универсальный тип для функций +F = TypeVar('F', bound=Callable[..., Any]) + +# Типы для aiogram событий +EventType = Union[Message, CallbackQuery] + + +class Logger: + """ + Кастомный логгер с поддержкой декораторов, прямого вызова и контекстных менеджеров. + + Features: + - Декораторы для sync/async функций + - Прямой вызов методов (debug, info, warning, error, critical) + - Автоматическое извлечение юзера из Message/CallbackQuery + - Раздельные файлы логов по уровням + - Ротация и retention логов + - Контекстные менеджеры для блоков кода + """ + + # Формат логов + _log_format: Final[str] = ( + '{time:YYYY-MM-DD HH:mm:ss.SSS} | ' + '{extra[system]}-{extra[log_type]} | ' + '{extra[user]} | {message}' + ) + + def __init__(self, system_name: str = 'PRIMO') -> None: + """ + Инициализация логгера. + + Args: + system_name: Имя системы для логирования (по умолчанию из settings) + """ + self.system_name = system_name or settings.PROJECT_NAME + self._setup_done = False + + def setup(self, start: bool = True) -> None: + """ + Настройка обработчиков Loguru: консоль и файлы. + + Учитывает переменную LOG_LEVEL из settings. + LOG_LEVEL определяет минимальный уровень для консоли и общего файла, + а также влияет на то, какие отдельные файлы создаются: + создаются только файлы для уровней >= LOG_LEVEL. + + Args: + start: Если True, сразу логирует запуск проекта + """ + if self._setup_done: + return + + # Полная очистка настроек + nlogger.remove() + + # Определяем уровень логирования из настроек + log_level_str = getattr(settings, 'LOG_LEVEL', 'INFO').upper() + # Проверка на допустимость + try: + log_level_no = nlogger.level(log_level_str).no + except ValueError: + log_level_str = 'INFO' + log_level_no = nlogger.level('INFO').no + + # Создание директории для файловых логов + log_dir: Path = settings.LOG_DIR + if not log_dir.exists(): + log_dir.mkdir(parents=True, exist_ok=True) + + # Консольный лог + if settings.LOG_CONSOLE: + nlogger.add( + sink=stderr, + format=self._log_format, + colorize=True, + level=log_level_str, + filter=lambda rec: rec['extra'].get('log_type') != 'TRACE' + ) + + # Файловые логи + if settings.LOG_FILE: + # Общий лог со всеми уровнями (начиная с LOG_LEVEL) + nlogger.add( + sink=log_dir / 'bot.log', + rotation=settings.LOG_ROTATION, + retention=settings.LOG_RETENTION, + format=self._log_format, + level=log_level_str, + enqueue=True, + backtrace=True, + diagnose=True, + encoding='utf-8' + ) + + # Раздельные логи по уровням – создаём только для уровней >= LOG_LEVEL + # Список интересующих нас уровней (в порядке возрастания) + level_configs = [ + ('DEBUG', 'debug.log'), + ('INFO', 'info.log'), + ('SUCCESS', 'success.log'), + ('WARNING', 'warning.log'), + ('ERROR', 'error.log'), + ('CRITICAL', 'critical.log') + ] + + for level_name, filename in level_configs: + level_no = nlogger.level(level_name).no + if level_no >= log_level_no: + nlogger.add( + sink=log_dir / filename, + rotation='10 MB', + retention=settings.LOG_RETENTION, + format=self._log_format, + level=level_name, + filter=lambda rec, lvl=level_name: rec['level'].name == lvl, + enqueue=True, + encoding='utf-8' + ) + + self._setup_done = True + + # Логируем старт + if start: + self.log_entry( + level='INFO', + text=f'Запуск проекта {self.system_name}...', + log_type='START' + ) + + @staticmethod + def format_user(event: Optional[EventType] = None) -> str: + """ + Форматирует имя пользователя из объекта Message или CallbackQuery. + + Args: + event: Объект Message или CallbackQuery + + Returns: + str: Строка '@username' или 'id' или '@System' + """ + if event is None: + return '@System' + + # Извлекаем пользователя из Message или CallbackQuery + user: Optional[User] = None + if isinstance(event, Message): + user = event.from_user + elif isinstance(event, CallbackQuery): + user = event.from_user + + if user is None: + return '@System' + + return f"@{user.username}" if user.username else f"id{user.id}" + + def log_entry( + self, + level: str, + text: str, + log_type: str = 'SYSTEM', + user: Optional[str] = None, + message: Optional[EventType] = None + ) -> None: + """ + Основной метод для записи логов. + + Args: + level: Уровень логирования ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL') + text: Сообщение для логирования + log_type: Кастомный тип лога (например, 'HANDLER', 'COMMAND', 'SPAM') + user: Явно указанный пользователь (если None, извлекается из message) + message: Объект Message/CallbackQuery для извлечения юзера + """ + actual_user: str = user or self.format_user(message) + nlogger.bind( + system=self.system_name, + user=actual_user, + log_type=log_type + ).log(level, text) + + def log( + self, + level: str = 'INFO', + log_type: str = 'HANDLER', + text: Optional[str] = None + ) -> Callable[[F], F]: + """ + Декоратор для логирования функций (sync и async). + + Args: + level: Уровень логирования + log_type: Категория лога + text: Кастомный текст сообщения (если None, используется имя функции) + + Returns: + Декорированную функцию + + Example: + ```python + @logger.log(level='INFO', log_type='COMMAND', text='Команда /start') + async def start_handler(message: Message): + await message.answer("Привет!") + ``` + """ + + def decorator(func: F) -> F: + is_coroutine = iscoroutinefunction(func) + action_text = text or f'Вызов {func.__name__}' + + @wraps(func) + def sync_wrapper(*args: Any, **kwargs: Any) -> Any: + message = self._find_event(args) + self.log_entry(level, f"[▶] {action_text}", log_type, message=message) + try: + result = func(*args, **kwargs) + self.log_entry(level, f"[✓] {action_text}", log_type, message=message) + return result + except Exception as e: + self.log_entry( + 'ERROR', + f"[✗] {action_text} | Exception: {e!r}", + log_type, + message=message + ) + raise + + @wraps(func) + async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + message = self._find_event(args) + self.log_entry(level, f"[▶] {action_text}", log_type, message=message) + try: + result = await func(*args, **kwargs) + self.log_entry(level, f"[✓] {action_text}", log_type, message=message) + return result + except Exception as e: + self.log_entry( + 'ERROR', + f"[✗] {action_text} | Exception: {e!r}", + log_type, + message=message + ) + raise + + return cast(F, async_wrapper if is_coroutine else sync_wrapper) + + return decorator + + @staticmethod + def _find_event(args: tuple[Any, ...]) -> Optional[EventType]: + """ + Ищет объект Message или CallbackQuery в аргументах функции. + + Args: + args: Аргументы функции + + Returns: + Найденный Message/CallbackQuery или None + """ + for arg in args: + if isinstance(arg, (Message, CallbackQuery)): + return arg + return None + + # ================= МЕТОДЫ ДЛЯ ПРЯМОГО ВЫЗОВА ================= + + def debug( + self, + text: str, + log_type: str = 'DEBUG', + user: Optional[str] = None, + message: Optional[EventType] = None + ) -> None: + """Логирование уровня DEBUG""" + self.log_entry('DEBUG', text, log_type, user, message) + + def info( + self, + text: str, + log_type: str = 'INFO', + user: Optional[str] = None, + message: Optional[EventType] = None + ) -> None: + """Логирование уровня INFO""" + self.log_entry('INFO', text, log_type, user, message) + + def warning( + self, + text: str, + log_type: str = 'WARNING', + user: Optional[str] = None, + message: Optional[EventType] = None + ) -> None: + """Логирование уровня WARNING""" + self.log_entry('WARNING', text, log_type, user, message) + + def error( + self, + text: str, + log_type: str = 'ERROR', + user: Optional[str] = None, + message: Optional[EventType] = None + ) -> None: + """Логирование уровня ERROR""" + self.log_entry('ERROR', text, log_type, user, message) + + def critical( + self, + text: str, + log_type: str = 'CRITICAL', + user: Optional[str] = None, + message: Optional[EventType] = None + ) -> None: + """Логирование уровня CRITICAL""" + self.log_entry('CRITICAL', text, log_type, user, message) + + def success( + self, + text: str, + log_type: str = 'SUCCESS', + user: Optional[str] = None, + message: Optional[EventType] = None + ) -> None: + """Логирование успешного выполнения (уровень SUCCESS)""" + self.log_entry('SUCCESS', text, log_type, user, message) + + # ================= КОНТЕКСТНЫЕ МЕНЕДЖЕРЫ ================= + + @contextmanager + def log_context( + self, + action: str, + log_type: str = 'CONTEXT', + level: str = 'INFO', + message: Optional[EventType] = None + ): + """ + Контекстный менеджер для логирования блока кода. + + Example: + ```python + with logger.log_context("Обработка данных", log_type='DATA'): + # ... ваш код ... + pass + ``` + """ + self.log_entry(level, f"[▶] {action}", log_type, message=message) + try: + yield + self.log_entry(level, f"[✓] {action}", log_type, message=message) + except Exception as e: + self.log_entry('ERROR', f"[✗] {action} | Exception: {e!r}", log_type, message=message) + raise + + +# ================= ГЛОБАЛЬНЫЙ ЭКЗЕМПЛЯР ================= + +# Создаем глобальный экземпляр логгера +logger: Logger = Logger(system_name="PRIMO") + +# Алиасы для обратной совместимости +setup_logging = logger.setup +log = logger.log