# [DEF:superset_tool.utils.logger:Module] # # @SEMANTICS: logging, utility, infrastructure, wrapper # @PURPOSE: Предоставляет универсальную обёртку над стандартным `logging.Logger` для унифицированного создания и управления логгерами с выводом в консоль и/или файл. # @LAYER: Infra # @RELATION: WRAPS -> logging.Logger # # @INVARIANT: Логгер всегда должен иметь имя. # @PUBLIC_API: SupersetLogger # [SECTION: IMPORTS] import logging import sys from datetime import datetime from pathlib import Path from typing import Optional, Any, Mapping # [/SECTION] # [DEF:SupersetLogger:Class] # @PURPOSE: Обёртка над `logging.Logger`, которая упрощает конфигурацию и использование логгеров. # @RELATION: WRAPS -> logging.Logger class SupersetLogger: # [DEF:SupersetLogger.__init__:Function] # @PURPOSE: Конфигурирует и инициализирует логгер, добавляя обработчики для файла и/или консоли. # @PRE: Если log_dir указан, путь должен быть валидным (или создаваемым). # @POST: `self.logger` готов к использованию с настроенными обработчиками. # @PARAM: name (str) - Идентификатор логгера. # @PARAM: log_dir (Optional[Path]) - Директория для сохранения лог-файлов. # @PARAM: level (int) - Уровень логирования (e.g., `logging.INFO`). # @PARAM: console (bool) - Флаг для включения вывода в консоль. def __init__(self, name: str = "superset_tool", log_dir: Optional[Path] = None, level: int = logging.INFO, console: bool = True, logger: Optional[logging.Logger] = None) -> None: if logger: self.logger = logger return self.logger = logging.getLogger(name) self.logger.setLevel(level) self.logger.propagate = False formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") if self.logger.hasHandlers(): self.logger.handlers.clear() if log_dir: log_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d") file_handler = logging.FileHandler(log_dir / f"{name}_{timestamp}.log", encoding="utf-8") file_handler.setFormatter(formatter) self.logger.addHandler(file_handler) if console: console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) self.logger.addHandler(console_handler) # [/DEF:SupersetLogger.__init__] # [DEF:SupersetLogger._log:Function] # @PURPOSE: (Helper) Универсальный метод для вызова соответствующего уровня логирования. # @PARAM: level_method (Any) - Метод логгера (info, debug, etc). # @PARAM: msg (str) - Сообщение. # @PARAM: args (Any) - Аргументы форматирования. # @PARAM: extra (Optional[Mapping[str, Any]]) - Дополнительные данные. # @PARAM: exc_info (bool) - Добавлять ли информацию об исключении. def _log(self, level_method: Any, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: level_method(msg, *args, extra=extra, exc_info=exc_info) # [/DEF:SupersetLogger._log] # [DEF:SupersetLogger.info:Function] # @PURPOSE: Записывает сообщение уровня INFO. def info(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: self._log(self.logger.info, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:SupersetLogger.info] # [DEF:SupersetLogger.debug:Function] # @PURPOSE: Записывает сообщение уровня DEBUG. def debug(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: self._log(self.logger.debug, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:SupersetLogger.debug] # [DEF:SupersetLogger.warning:Function] # @PURPOSE: Записывает сообщение уровня WARNING. def warning(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: self._log(self.logger.warning, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:SupersetLogger.warning] # [DEF:SupersetLogger.error:Function] # @PURPOSE: Записывает сообщение уровня ERROR. def error(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: self._log(self.logger.error, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:SupersetLogger.error] # [DEF:SupersetLogger.critical:Function] # @PURPOSE: Записывает сообщение уровня CRITICAL. def critical(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: self._log(self.logger.critical, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:SupersetLogger.critical] # [DEF:SupersetLogger.exception:Function] # @PURPOSE: Записывает сообщение уровня ERROR вместе с трассировкой стека текущего исключения. def exception(self, msg: str, *args: Any, **kwargs: Any) -> None: self.logger.exception(msg, *args, **kwargs) # [/DEF:SupersetLogger.exception] # [/DEF:SupersetLogger] # [/DEF:superset_tool.utils.logger]