ss-tools/superset_tool/utils/logger.py

# [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]