Почему логер для INFO использует Formatter для Debug и почему дублируются сообщения логера

Сам немного экспериментировал с логгером, надеюсь мой пример вам окажется полезным. Добивал специально для вас комментариев и примеры использования:

import platform
import os
import inspect
import logging
import logging.handlers
from datetime import datetime
from typing import List, Optional

# Дополнительная проверка для ОС Windows
if platform.system() == "Windows":
    os.system('color')

class CustomFormatter(logging.Formatter):
    DEFAULT_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)"

    COLORS = {
        "black": "\x1b[30m",
        "red": "\x1b[31m",
        "green": "\x1b[32m",
        "yellow": "\x1b[33m",
        "blue": "\x1b[34m",
        "magenta": "\x1b[35m",
        "cyan": "\x1b[36m",
        "white": "\x1b[37m",
        "reset": "\x1b[0m"
    }

    LEVEL_COLORS = {
        logging.DEBUG: COLORS["green"],
        logging.INFO: COLORS["magenta"],
        logging.WARNING: COLORS["yellow"],
        logging.ERROR: COLORS["red"],
        logging.CRITICAL: COLORS["cyan"]
    }

    CONSOLE_FORMATS = {
        logging.DEBUG: LEVEL_COLORS[logging.DEBUG] + DEFAULT_FORMAT + COLORS["reset"],
        logging.INFO: LEVEL_COLORS[logging.INFO] + DEFAULT_FORMAT + COLORS["reset"],
        logging.WARNING: LEVEL_COLORS[logging.WARNING] + DEFAULT_FORMAT + COLORS["reset"],
        logging.ERROR: LEVEL_COLORS[logging.ERROR] + DEFAULT_FORMAT + COLORS["reset"],
        logging.CRITICAL: LEVEL_COLORS[logging.CRITICAL] + DEFAULT_FORMAT + COLORS["reset"]
    }

    FILE_FORMATS = {
        logging.DEBUG: DEFAULT_FORMAT,
        logging.INFO: DEFAULT_FORMAT,
        logging.WARNING: DEFAULT_FORMAT,
        logging.ERROR: DEFAULT_FORMAT,
        logging.CRITICAL: DEFAULT_FORMAT
    }

    def __init__(self, is_file=False):
        super().__init__()
        self.is_file = is_file

    def format(self, record: logging.LogRecord) -> str:
        if self.is_file:
            log_fmt = self.FILE_FORMATS.get(record.levelno, self.DEFAULT_FORMAT)
        else:
            log_fmt = self.CONSOLE_FORMATS.get(record.levelno, self.DEFAULT_FORMAT)
        self._style._fmt = log_fmt
        return super().format(record)

    @classmethod
    def set_file_format(cls, level: str, format_string: str):
        """Устанавливает форматирование для заданного уровня логирования для файлового вывода."""
        log_level = getattr(logging, level.upper(), None)
        if log_level is not None:
            cls.FILE_FORMATS[log_level] = format_string
        else:
            raise ValueError(f"Некорректный уровень логирования: {level}")

    @classmethod
    def set_console_format(cls, level: str, format_string: str):
        """Устанавливает форматирование для заданного уровня логирования."""
        log_level = getattr(logging, level.upper(), None)
        if log_level is not None:
            cls.CONSOLE_FORMATS[log_level] = format_string
        else:
            raise ValueError(f"Некорректный уровень логирования: {level}")

    @classmethod
    def set_color(cls, level: str, color: str):
        """Устанавливает цвет для заданного уровня логирования."""
        log_level = getattr(logging, level.upper(), None)
        if log_level in cls.LEVEL_COLORS:
            cls.LEVEL_COLORS[log_level] = cls.COLORS[color]
            cls.CONSOLE_FORMATS[log_level] = cls.LEVEL_COLORS[log_level] + cls.DEFAULT_FORMAT + cls.COLORS["reset"]
        else:
            raise ValueError(f"Некорректный уровень логирования: {level}")


class MultiLevelFilter(logging.Filter):
    def __init__(self, levels: List[int]):
        super().__init__()
        self.levels = levels

    def filter(self, record: logging.LogRecord) -> bool:
        return record.levelno in self.levels


class LevelFilter(logging.Filter):
    def __init__(self, level: int):
        super().__init__()
        self.level = level

    def filter(self, record: logging.LogRecord) -> bool:
        return record.levelno == self.level


class LoggerManager:
    LEVEL_MAPPING = {
        'DEBUG': logging.DEBUG,
        'INFO': logging.INFO,
        'WARNING': logging.WARNING,
        'ERROR': logging.ERROR,
        'CRITICAL': logging.CRITICAL
    }

    def __init__(self, *, name: str, level: int = logging.DEBUG, log_to_file: bool = False, file_name: str = 'app.log',
                 log_dir: str = 'logs', max_bytes: int = 1024 * 1024, backup_count: int = 5):
        if not isinstance(name, str) or not name.strip():
            print(f"Некорректное имя логгера: '{name}'. Устанавливается имя по умолчанию 'DefaultLogger'.")
            name = "DefaultLogger"
        self.max_bytes: int = max_bytes
        self.backup_count: int = backup_count
        self.logger = logging.getLogger(name)
        self.logger.setLevel(level)
        self.console_handler: Optional[logging.StreamHandler] = None
        self.file_handler: Optional[logging.FileHandler] = None
        self.log_filter: Optional[logging.Filter] = None
        self.setup_console_handler(level)
        if log_to_file:
            self.enable_file_logging(file_name, log_dir)

    def setup_console_handler(self, level: int):
        """Настройка консольного обработчика и форматтера"""
        self.console_handler = logging.StreamHandler()
        self.console_handler.setLevel(level)
        self.console_handler.setFormatter(CustomFormatter())
        self.logger.addHandler(self.console_handler)

    def enable_console_logging(self):
        """Включение логирования в консоль"""
        if self.console_handler:
            self.console_handler.setLevel(self.logger.level)
            self.logger.info("Логирование в консоль включено")

    def disable_console_logging(self):
        """Отключение логирования в консоль"""
        if self.console_handler:
            self.console_handler.setLevel(
                logging.CRITICAL + 1)  # Уровень выше CRITICAL, чтобы игнорировать все сообщения
            self.logger.info("Логирование в консоль отключено")
        else:
            self.logger.warning("Обработчик консольного логирования уже отключен или не был инициализирован")

    def enable_logging(self):
        """Включение логирования"""
        self.logger.disabled = False
        self.logger.info("Логирование включено")

    def disable_logging(self):
        """Отключение логирования"""
        self.logger.disabled = True
        self.logger.info("Логирование отключено")

    def enable_file_logging(self, file_name: str = 'app.log', log_dir: str = 'logs'):
        try:
            if not os.path.exists(log_dir):
                os.makedirs(log_dir)
        except Exception as e:
            self.logger.error(f"Ошибка при создании директории для логов: {e}")
            raise

        log_path = os.path.join(log_dir, file_name)

        if self.file_handler:
            self.logger.removeHandler(self.file_handler)
            self.file_handler.close()

        try:
            self.file_handler = logging.handlers.RotatingFileHandler(
                log_path,
                maxBytes=self.max_bytes,  # Максимальный размер файла
                backupCount=self.backup_count  # Количество ротируемых файлов
            )
            self.file_handler.setLevel(self.logger.level)
            file_formatter = CustomFormatter(is_file=True)
            self.file_handler.setFormatter(file_formatter)
            self.logger.addHandler(self.file_handler)
            self.logger.info(f"Логирование в файл '{log_path}' включено")
        except Exception as e:
            self.logger.error(f"Ошибка при настройке файлового логгера с ротацией: {e}")
            raise

    def disable_file_logging(self):
        """Отключение логирования в файл"""
        if self.file_handler:
            self.logger.removeHandler(self.file_handler)
            self.file_handler.close()
            self.file_handler = None
            self.logger.info("Логирование в файл отключено")
        else:
            self.logger.warning("Обработчик файла логирования уже отключен или не был инициализирован")

    def set_file_handler_params(self, max_bytes: int, backup_count: int):
        """Установка параметров для RotatingFileHandler."""
        self.max_bytes = max_bytes
        self.backup_count = backup_count
        if self.file_handler:
            self.file_handler.maxBytes = max_bytes
            self.file_handler.backupCount = backup_count
            self.logger.info(
                f"Изменены параметры RotatingFileHandler: maxBytes={max_bytes}, backupCount={backup_count}")
        else:
            self.logger.warning(
                "Файловый обработчик не инициализирован. Новые параметры будут применены при следующем включении логирования в файл.")

    def set_name(self, name: str):
        """Установка нового имени логгера"""
        if not isinstance(name, str) or not name.strip():
            self.logger.error(f"Некорректное новое имя логгера: '{name}'. Имя остается без изменений.")
            return

        self.logger.name = name
        self.logger.info(f"Имя логгера изменено на '{name}'")

    def set_level(self, level: str):
        """Установка уровня логирования"""
        log_level = self.LEVEL_MAPPING.get(level)
        if log_level is not None:
            self.logger.setLevel(log_level)
            if self.console_handler:
                self.console_handler.setLevel(log_level)
            if self.file_handler:
                self.file_handler.setLevel(log_level)
            self.logger.info(f"Уровень логирования установлен на {level}")
        else:
            self.logger.error(
                f"Некорректный уровень логирования: {level}. Используйте один из: {list(self.LEVEL_MAPPING.keys())}")

    def set_filter(self, level: str):
        """Установка фильтра для логирования"""
        log_level = self.LEVEL_MAPPING.get(level)
        if log_level is not None:
            self.clear_filter()
            self.log_filter = LevelFilter(log_level)
            self.logger.addFilter(self.log_filter)
            self.logger.info(f"Фильтр уровня логирования установлен на {level}")
        else:
            self.logger.error(
                f"Некорректный уровень фильтра: {level}. Используйте один из: {list(self.LEVEL_MAPPING.keys())}")

    def set_filter_list(self, levels: List[str]):
        """Установка списка фильтров для логирования"""
        if not levels:
            self.logger.error("Список уровней фильтра не должен быть пустым.")
            return

        valid_levels = [self.LEVEL_MAPPING[level] for level in levels if level in self.LEVEL_MAPPING]

        if not valid_levels:
            self.logger.error(
                f"Некорректные уровни фильтра: {levels}. Используйте один из: {list(self.LEVEL_MAPPING.keys())}")
            return

        self.clear_filter()
        self.log_filter = MultiLevelFilter(valid_levels)
        self.logger.addFilter(self.log_filter)
        self.logger.info(f"Фильтр уровней логирования установлен на {levels}")

    def reset_level(self):
        """Сброс уровня логирования до уровня DEBUG"""
        self.set_level('DEBUG')

    def clear_filter(self):
        """Очистка установленного фильтра"""
        if self.log_filter:
            self.logger.removeFilter(self.log_filter)
            self.log_filter = None
            self.logger.info("Фильтр логирования очищен")

    def time_execution(self):
        """Контекстный менеджер для логирования времени выполнения блока кода"""
        return TimeExecutionLogger(self.logger)


class TimeExecutionLogger:
    def __init__(self, with_logger, level=logging.INFO):
        self.logger = with_logger
        self.level = level
        self.start_time = datetime.now()
        caller_frame = inspect.stack()[
            2]  # Получаем информацию о вызывающем коде (2 - это индекс фрейма с вызывающим кодом)
        self.caller_file = caller_frame.filename
        self.caller_line = caller_frame.lineno
        self.logger.log(self.level,
                        f"Начало выполнения блока кода в {self.start_time.strftime('%Y-%m-%d %H:%M:%S.%f')}. Вызвано из файла '{self.caller_file}', строка {self.caller_line}")

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        end_time = datetime.now()
        execution_time = (end_time - self.start_time).total_seconds()
        if exc_type is None:
            self.logger.log(self.level,
                            f"Завершение выполнения блока кода в {end_time.strftime('%Y-%m-%d %H:%M:%S.%f')}. Время выполнения: {execution_time:.6f} секунд. Вызвано из файла '{self.caller_file}', строка {self.caller_line}")
        else:
            self.logger.log(self.level,
                            f"Завершение выполнения блока кода с ошибкой {exc_val} в {end_time.strftime('%Y-%m-%d %H:%M:%S.%f')}. Время выполнения: {execution_time:.6f} секунд. Вызвано из файла '{self.caller_file}', строка {self.caller_line}")
        return False  # Возвращаем False, чтобы исключение продолжило подъём


if __name__ == "__main__":
    import time

    logger_manager = LoggerManager(name="TestLogger")
    # Установим форматы для файлового логирования
    CustomFormatter.set_file_format('DEBUG', "%(asctime)s - !DEBUG!: %(message)s (%(filename)s:%(lineno)d)")
    CustomFormatter.set_file_format('INFO', "%(asctime)s - !INFO!: %(message)s")
    CustomFormatter.set_file_format('WARNING', "%(asctime)s - !WARNING!: %(message)s")
    CustomFormatter.set_file_format('ERROR', "%(asctime)s - !ERROR!: %(message)s")
    # CustomFormatter.set_file_format('CRITICAL', "%(asctime)s - !CRITICAL!: %(message)s")
    # CustomFormatter.set_console_format('INFO', f"%(asctime)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)") # Задать свой формат
    # CustomFormatter.set_color('DEBUG', 'blue')  # Задать цвет из доступных

    # logger_manager.set_file_handler_params(max_bytes = 2048*1024, backup_count = 10) # Максимальный размер файла и количество ротируемых файлов
    # logger_manager.enable_console_logging() # Включение логирования в консоль
    # logger_manager.disable_console_logging() # Отключение логирования в консоль
    # logger_manager.disable_logging()  # Отключение логирования
    # logger_manager.enable_logging()   # Включение логирования
    # logger_manager.set_level('INFO')  # Установка уровня логирования на INFO
    # logger_manager.set_filter_list(['DEBUG' ,'WARNING', 'ERROR' ]) # Установка списка фильтров для логирования
    # logger_manager.set_filter('INFO')  # Фильтрация сообщений только уровня INFO
    # logger_manager.clear_filter()  # Очистка установленного фильтра
    # logger_manager.reset_level()  # Сброс уровня логирования до уровня DEBUG
    # logger_manager.set_name("NewNameLogger")  # Установка нового имени логгера
    with logger_manager.time_execution():
        for i in range(6):
            # Включение и отключение логирования в файл
            logger_manager.enable_file_logging(f"{datetime.now().strftime('%Y-%m-%d %H-%M')}-{i}.log")
            # logger_manager.disable_file_logging()

            # Логирование сообщений
            logger = logger_manager.logger
            logger.debug(f"debug message {i}")
            logger.info(f"info message {i}")
            logger.warning(f"warning message {i}")
            logger.error(f"error message {i}")
            logger.critical(f"critical message {i}")
            time.sleep(1)

Спасибо огромное за пример. Добавил к вашему коду переопределение дебагера с добавлением переменных:

class MyLogger(logging.Logger):
    def __init__(self, name, level=logging.INFO):
        super().__init__(name, level)

    def debug(self, msg, *args, data=None, **kwargs):

        current_frame = inspect.currentframe()
        caller_frame = current_frame.f_back
        code_obj = caller_frame.f_code
        code_obj_name = code_obj.co_name
        line = caller_frame.f_lineno
        full_file_name = code_obj.co_filename
        filename = full_file_name[full_file_name.rfind('\\') +1:]
        data_module = {'full_file_name': full_file_name, 'code_obj_name': code_obj_name, 'line': line}

        if data:
            print(data)

        super().debug(msg, *args, extra=data_module,  **kwargs)

В class LoggerManager: изменил self.logger = logging.getLogger(name) на self.logger = MyLogger(name).

Добавил дополнительный формат для переопределенного debug:

DEFAULT_FORMAT = "%(asctime)s - [%(levelname)s] - %(name)s - (%(filename)s).%(funcName)s(%(lineno)d) - %(message)s"
DEBUG_FORMAT = '%(asctime)s - [%(levelname)s] - %(name)s - (%(full_file_name)s).%(code_obj_name)s(%(line)d) - %(message)s'

Иначе в лог прилетают данные из самого класса. Теперь работает как нужно. Можно в debug дополнительно передавать необходимые данные. И в логе корректно отображается откуда был вызван логер и из какой строки. Спасибо большое за пример!