# # @SEMANTICS: logging, utility, infrastructure, wrapper # @PURPOSE: Предоставляет универсальную обёртку над стандартным `logging.Logger` для унифицированного создания и управления логгерами с выводом в консоль и/или файл. # import logging import sys from datetime import datetime from pathlib import Path from typing import Optional, Any, Mapping # # --- Начало кода модуля --- # # @PURPOSE: Обёртка над `logging.Logger`, которая упрощает конфигурацию и использование логгеров. # @RELATION: WRAPS -> logging.Logger class SupersetLogger: def __init__(self, name: str = "superset_tool", log_dir: Optional[Path] = None, level: int = logging.INFO, console: bool = True) -> None: # # @PURPOSE: Конфигурирует и инициализирует логгер, добавляя обработчики для файла и/или консоли. # @PARAM: name: str - Идентификатор логгера. # @PARAM: log_dir: Optional[Path] - Директория для сохранения лог-файлов. # @PARAM: level: int - Уровень логирования (e.g., `logging.INFO`). # @PARAM: console: bool - Флаг для включения вывода в консоль. # @POST: `self.logger` готов к использованию с настроенными обработчиками. 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) # # # @PURPOSE: (Helper) Универсальный метод для вызова соответствующего уровня логирования. # @INTERNAL 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) # # # @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) # # # @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) # # # @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) # # # @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) # # # @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) # # # @PURPOSE: Записывает сообщение уровня ERROR вместе с трассировкой стека текущего исключения. def exception(self, msg: str, *args: Any, **kwargs: Any) -> None: self.logger.exception(msg, *args, **kwargs) # # # --- Конец кода модуля --- #