logmanager/logmanager.py - master - basswan/py_bot

logmanager/logmanager.py
89 строк3 KB
basswan
create: logmanager.py, abstractcommand.py, executor.py, invoke.py, main.py
09 апр 2026, 15:59
09 апр 2026, 15:59821d5d8
import logging
from logging.handlers import RotatingFileHandler
from typing import Dict

# Константы по умолчанию
_DEFAULT_LOG_FORMAT = "%(asctime)s - [%(levelname)s] - %(name)s - (%(filename)s).%(funcName)s(%(lineno)d) - %(message)s"
_DEFAULT_FILE_SIZE = 20 * 1024 * 1024  # 20 MB
_DEFAULT_BACKUP_COUNT = 15


class LogManager:
    """
    Синглтон-менеджер для централизованного логирования с ротацией файлов.
    """

    def __init__(self):
        self._file_name: str = ""
        self._file_size: int = _DEFAULT_FILE_SIZE
        self._backup_count: int = _DEFAULT_BACKUP_COUNT
        self._loggers: Dict[str, logging.Logger] = {}

    def configure(self, file_name: str, file_size: int = _DEFAULT_FILE_SIZE, backup_count: int = _DEFAULT_BACKUP_COUNT):
        """
        Настраивает путь к файлу и параметры ротации.
        Должен быть вызван до первого использования логгера.
        """
        if not file_name:
            raise ValueError("file_name must be specified")
        self._file_name = file_name
        self._file_size = file_size
        self._backup_count = backup_count

    def get_logger(self, name: str) -> logging.Logger:
        if name in self._loggers:
            return self._loggers[name]

        logger = logging.getLogger(name)
        logger.setLevel(logging.INFO)

        # Предотвращаем дублирование логов
        logger.propagate = False

        handler = RotatingFileHandler(
            self._file_name,
            maxBytes=self._file_size,
            backupCount=self._backup_count
        )
        formatter = logging.Formatter(_DEFAULT_LOG_FORMAT)
        handler.setFormatter(formatter)
        logger.addHandler(handler)

        self._loggers[name] = logger
        return logger

    def add_info(self, msg: str, name: str) -> None:
        self.get_logger(name).info(msg)

    def add_warning(self, msg: str, name: str) -> None:
        self.get_logger(name).warning(msg)

    def add_error(self, msg: str, name: str) -> None:
        self.get_logger(name).error(msg)

    def add_log(self, msg: str, name: str = "COMPAT") -> None:
        """Совместимость со старым API."""
        self.add_info(msg, name)


# Единый экземпляр
_log_manager = None


def _get_log_manager() -> LogManager:
    global _log_manager
    if _log_manager is None:
        _log_manager = LogManager()
    return _log_manager


def add_info_to_log(msg: str, name: str) -> None:
    _get_log_manager().add_info(msg, name)


def add_warning_to_log(msg: str, name: str) -> None:
    _get_log_manager().add_warning(msg, name)


def add_error_to_log(msg: str, name: str) -> None:
    _get_log_manager().add_error(msg, name)