ss-tools/superset_tool/utils/logger.py

# [MODULE_PATH] superset_tool.utils.logger
# [FILE] logger.py
# [SEMANTICS] logging, utils, ai‑friendly, infrastructure

# --------------------------------------------------------------
# [IMPORTS]
# --------------------------------------------------------------
import logging
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional, Any, Mapping
# [END_IMPORTS]

# --------------------------------------------------------------
# [ENTITY: Service('SupersetLogger')]
# --------------------------------------------------------------
"""
:purpose: Универсальная обёртка над ``logging.Logger``. Позволяет:
          • задавать уровень и вывод в консоль/файл,
          • передавать произвольные ``extra``‑поля,
          • использовать привычный API (info, debug, warning, error,
            critical, exception) без «падения» при неверных аргументах.
:preconditions:
    - ``name`` – строка‑идентификатор логгера,
    - ``level`` – валидный уровень из ``logging``,
    - ``log_dir`` – при указании директория, куда будет писаться файл‑лог.
:postconditions:
    - Создан полностью сконфигурированный ``logging.Logger`` без
      дублирующих обработчиков.
"""
class SupersetLogger:
    """
    :ivar logging.Logger logger: Внутренний стандартный логгер.
    :ivar bool propagate: Отключаем наследование записей, чтобы
                          сообщения не «проваливались» выше.
    """

    # --------------------------------------------------------------
    # [ENTITY: Method('__init__')]
    # --------------------------------------------------------------
    """
    :purpose: Конфигурировать базовый логгер, добавить обработчики
              консоли и/или файла, очистить прежние обработчики.
    :preconditions: Параметры валидны.
    :postconditions: ``self.logger`` готов к использованию.
    """
    def __init__(
        self,
        name: str = "superset_tool",
        log_dir: Optional[Path] = None,
        level: int = logging.INFO,
        console: bool = True,
    ) -> None:
        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)

    # [END_ENTITY]

    # --------------------------------------------------------------
    # [ENTITY: Method('_log')]
    # --------------------------------------------------------------
    """
    :purpose: Универсальная вспомогательная обёртка над
              ``logging.Logger.<level>``. Принимает любые ``*args``
              (подстановочные параметры) и ``extra``‑словарь.
    :preconditions:
        - ``level_method`` – один из методов ``logger``,
        - ``msg`` – строка‑шаблон,
        - ``*args`` – значения для ``%``‑подстановок,
        - ``extra`` – пользовательские атрибуты (может быть ``None``).
    :postconditions: Запись в журнал выполнена.
    """
    def _log(
        self,
        level_method: Any,
        msg: str,
        *args: Any,
        extra: Optional[Mapping[str, Any]] = None,
        exc_info: bool = False,
    ) -> None:
        if extra is not None:
            level_method(msg, *args, extra=extra, exc_info=exc_info)
        else:
            level_method(msg, *args, exc_info=exc_info)

    # [END_ENTITY]

    # --------------------------------------------------------------
    # [ENTITY: Method('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)
    # [END_ENTITY]

    # --------------------------------------------------------------
    # [ENTITY: Method('debug')]
    # --------------------------------------------------------------
    """
    :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)
    # [END_ENTITY]

    # --------------------------------------------------------------
    # [ENTITY: Method('warning')]
    # --------------------------------------------------------------
    """
    :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)
    # [END_ENTITY]

    # --------------------------------------------------------------
    # [ENTITY: Method('error')]
    # --------------------------------------------------------------
    """
    :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)
    # [END_ENTITY]

    # --------------------------------------------------------------
    # [ENTITY: Method('critical')]
    # --------------------------------------------------------------
    """
    :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)
    # [END_ENTITY]

    # --------------------------------------------------------------
    # [ENTITY: Method('exception')]
    # --------------------------------------------------------------
    """
    :purpose: Записать сообщение уровня ERROR вместе с трассировкой
              текущего исключения (аналог ``logger.exception``).
    """
    def exception(self, msg: str, *args: Any, **kwargs: Any) -> None:
        self.logger.exception(msg, *args, **kwargs)
    # [END_ENTITY]

# --------------------------------------------------------------
# [END_FILE logger.py]
# --------------------------------------------------------------