Исходный код core.configs.log

"""Настройка логгеров приложения и вывода внешних скриптов."""

import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path

from pydantic import BaseModel




class LoggingSettings(BaseModel):
    """Настройка логгеров: консоль плюс файл с ротацией."""
    FORMAT: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    LEVEL: str = "INFO"
    SCRIPT_LEVEL: str = "INFO"
    FLOWGRAPH_LEVEL: str = "INFO"
    DIRECTORY: str = "/var/lib/soniks-client/logs"
    FILE_NAME: str = "client.log"
    MAX_BYTES: int = 5 * 1024 * 1024  # 5 MB
    BACKUP_COUNT: int = 5
    ENCODING: str = "utf-8"

    @property
    def script_log_level(self) -> int:
        return logging._nameToLevel[self.SCRIPT_LEVEL]

    @property
    def flowgraph_log_level(self) -> int:
        return logging._nameToLevel[self.FLOWGRAPH_LEVEL]



    def configure_logger(self, name: str = "app") -> logging.Logger:
        """Собрать основной логгер приложения."""
        formatter = logging.Formatter(self.FORMAT)
        handlers = self._create_handlers(formatter)

        logger = logging.getLogger(name)
        logger.setLevel(self.LEVEL)
        logger.handlers.clear()

        for handler in handlers:
            logger.addHandler(handler)

        return logger




    def configure_raw_logger(self, name: str = "row_logs") -> logging.Logger:
        """Собрать логгер сырого вывода внешних скриптов (без префиксов)."""
        raw_formatter = logging.Formatter("%(message)s")
        handlers = self._create_handlers(raw_formatter)

        raw_logger = logging.getLogger(name)
        raw_logger.setLevel(self.SCRIPT_LEVEL)
        raw_logger.handlers.clear()

        for handler in handlers:
            raw_logger.addHandler(handler)

        raw_logger.propagate = False

        return raw_logger


    def _create_handlers(
        self,
        formatter: logging.Formatter,
    ) -> tuple[logging.Handler, logging.Handler]:
        Path(self.DIRECTORY).mkdir(parents=True, exist_ok=True)

        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)

        file_handler = RotatingFileHandler(
            filename=self._full_path,
            maxBytes=self.MAX_BYTES,
            backupCount=self.BACKUP_COUNT,
            encoding=self.ENCODING,
        )
        file_handler.setFormatter(formatter)

        return console_handler, file_handler

    @property
    def _full_path(self) -> Path:
        return Path(self.DIRECTORY) / self.FILE_NAME