# [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, Generator from contextlib import contextmanager # [/SECTION] # [DEF:belief_scope:Function] # @PURPOSE: Context manager for belief state logging to maintain execution coherence. # @PRE: scope_id must be a string. # @POST: Entry and exit actions are logged. # @PARAM: scope_id (str) - Identifier for the logical scope. @contextmanager def belief_scope(scope_id: str) -> Generator[None, None, None]: """Context manager for belief state logging.""" logger = logging.getLogger("superset_tool") logger.debug(f"[BELIEF_ENTRY] {scope_id}") try: yield finally: logger.debug(f"[BELIEF_EXIT] {scope_id}") # [/DEF:belief_scope:Function] # [DEF:SupersetLogger:Class] # @PURPOSE: Обёртка над `logging.Logger`, которая упрощает конфигурацию и использование логгеров. # @RELATION: WRAPS -> logging.Logger class SupersetLogger: # [DEF:__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: with belief_scope("__init__"): 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:__init__:Function] # [DEF:_log:Function] # @PURPOSE: (Helper) Универсальный метод для вызова соответствующего уровня логирования. # @PRE: level_method должен быть вызываемым методом логгера. msg must be a string. # @POST: Сообщение записано в лог. # @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: with belief_scope("_log"): level_method(msg, *args, extra=extra, exc_info=exc_info) # [/DEF:_log:Function] # [DEF:info:Function] # @PURPOSE: Записывает сообщение уровня INFO. # @PRE: msg должен быть строкой. # @POST: Сообщение уровня INFO записано. def info(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: with belief_scope("info"): self._log(self.logger.info, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:info:Function] # [DEF:debug:Function] # @PURPOSE: Записывает сообщение уровня DEBUG. # @PRE: msg должен быть строкой. # @POST: Сообщение уровня DEBUG записано. def debug(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: with belief_scope("debug"): self._log(self.logger.debug, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:debug:Function] # [DEF:warning:Function] # @PURPOSE: Записывает сообщение уровня WARNING. # @PRE: msg должен быть строкой. # @POST: Сообщение уровня WARNING записано. def warning(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: with belief_scope("warning"): self._log(self.logger.warning, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:warning:Function] # [DEF:error:Function] # @PURPOSE: Записывает сообщение уровня ERROR. # @PRE: msg должен быть строкой. # @POST: Сообщение уровня ERROR записано. def error(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: with belief_scope("error"): self._log(self.logger.error, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:error:Function] # [DEF:critical:Function] # @PURPOSE: Записывает сообщение уровня CRITICAL. # @PRE: msg должен быть строкой. # @POST: Сообщение уровня CRITICAL записано. def critical(self, msg: str, *args: Any, extra: Optional[Mapping[str, Any]] = None, exc_info: bool = False) -> None: with belief_scope("critical"): self._log(self.logger.critical, msg, *args, extra=extra, exc_info=exc_info) # [/DEF:critical:Function] # [DEF:exception:Function] # @PURPOSE: Записывает сообщение уровня ERROR вместе с трассировкой стека текущего исключения. # @PRE: msg должен быть строкой. # @POST: Сообщение об ошибке с traceback записано. def exception(self, msg: str, *args: Any, **kwargs: Any) -> None: with belief_scope("exception"): self.logger.exception(msg, *args, **kwargs) # [/DEF:exception:Function] # [DEF:belief_scope:Method] # @PURPOSE: Instance method wrapper for belief_scope context manager. # @PRE: scope_id must be a string. # @POST: Enters the belief scope. @contextmanager def belief_scope(self, scope_id: str) -> Generator[None, None, None]: with belief_scope(scope_id): yield # [/DEF:belief_scope:Method] # [/DEF:SupersetLogger:Class] # [/DEF:superset_tool.utils.logger:Module]