core/utils/logger.py

import logging
from pathlib import Path
from typing import Any

import colorlog
from graphql import GraphQLError

graphql_logger = logging.getLogger("graphql")
graphql_logger.setLevel(logging.WARNING)

ariadne_logger = logging.getLogger("ariadne")
ariadne_logger.setLevel(logging.WARNING)

_lib_path = Path(__file__).parents[1]
_leng_path = len(_lib_path.as_posix())


def console_filter(record: logging.LogRecord) -> bool:
    """🔍 Фильтр для консольного вывода - подавляет спам авторизации"""
    # Define `package` attribute with the relative path.
    record.package = record.pathname[_leng_path + 1 :].replace(".py", "")
    record.emoji = (
        "🔍"
        if record.levelno == logging.DEBUG
        else "ℹ︎"
        if record.levelno == logging.INFO
        else "🚧"
        if record.levelno == logging.WARNING
        else "❌"
        if record.levelno == logging.ERROR
        else "🧨"
        if record.levelno == logging.CRITICAL
        else ""
    )

    # 🚫 Подавляем логи ошибок авторизации ТОЛЬКО в консоли (не в Sentry)
    if record.levelno >= logging.ERROR and record.getMessage():
        message = record.getMessage()
        # Подавляем ошибки авторизации в консоли
        if any(
            phrase in message
            for phrase in [
                "Требуется авторизация",
                "AuthorizationError",
                "GraphQL request:",
                "Traceback (most recent call last):",
                "graphql.error.graphql_error.GraphQLError: Требуется авторизация",
                "load_drafts",
                "GetUserDocuments",
                "GetDrafts",
                "During handling of the above exception",
                'File "/usr/local/lib/python3.13/site-packages/graphql/execution/execute.py"',
                'File "/app/services/auth.py"',
                "await_result",
                "handle_field_error",
                "await gather",
                "get_results",
                "raise error",
                "return_type, field_nodes, info, path, await result",
                'raise AuthorizationError("Требуется авторизация")',
                "decorated_function",
            ]
        ):
            return False  # Не показываем в консоли ошибки авторизации

    # Подавляем повторяющиеся Ariadne логи в консоли
    if record.name in ["ariadne", "graphql"]:
        message = record.getMessage()
        if any(
            phrase in message
            for phrase in [
                "Требуется авторизация",
                "AuthorizationError",
                "load_drafts",
                "GraphQL request:",
            ]
        ):
            return False

    return True


def basic_filter(record: logging.LogRecord) -> bool:
    """🔍 Базовый фильтр для всех логов - только добавляет метаданные"""
    # Define `package` attribute with the relative path.
    record.package = record.pathname[_leng_path + 1 :].replace(".py", "")
    record.emoji = (
        "🔍"
        if record.levelno == logging.DEBUG
        else "ℹ︎"
        if record.levelno == logging.INFO
        else "🚧"
        if record.levelno == logging.WARNING
        else "❌"
        if record.levelno == logging.ERROR
        else "🧨"
        if record.levelno == logging.CRITICAL
        else ""
    )
    return True


# Define the color scheme
color_scheme = {
    "DEBUG": "light_black",
    "INFO": "green",
    "WARNING": "yellow",
    "ERROR": "red",
    "CRITICAL": "red,bg_white",
}

# Define secondary log colors
secondary_colors = {
    "log_name": {"DEBUG": "blue"},
    "asctime": {"DEBUG": "cyan"},
    "process": {"DEBUG": "purple"},
    "module": {"DEBUG": "light_black,bg_blue"},
    "funcName": {"DEBUG": "light_white,bg_blue"},  # Add this line
}

# Define the log format string
fmt_string = "%(emoji)s%(log_color)s%(package)s.%(funcName)s%(reset)s %(white)s%(message)s"

# Define formatting configuration
fmt_config = {
    "log_colors": color_scheme,
    "secondary_log_colors": secondary_colors,
    "style": "%",
    "reset": True,
}


class MultilineColoredFormatter(colorlog.ColoredFormatter):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.log_colors = kwargs.pop("log_colors", {})
        self.secondary_log_colors = kwargs.pop("secondary_log_colors", {})

    def format(self, record: logging.LogRecord) -> str:
        # Add default emoji if not present
        if not hasattr(record, "emoji"):
            record.emoji = "📝"

        # Add default package if not present
        if not hasattr(record, "package"):
            record.package = getattr(record, "name", "unknown")

        # Format the first line normally
        formatted_first_line = super().format(record)

        # Check if the message has multiple lines
        lines = formatted_first_line.split("\n")
        if len(lines) > 1:
            # For multiple lines, only apply colors to the first line
            # Keep subsequent lines without color formatting
            formatted_lines = [formatted_first_line]
            formatted_lines.extend(lines[1:])
            return "\n".join(formatted_lines)
        return super().format(record)


# Create a MultilineColoredFormatter object for colorized logging
formatter = MultilineColoredFormatter(fmt_string, **fmt_config)

# Create a stream handler for logging output (только для консоли)
stream = logging.StreamHandler()
stream.setFormatter(formatter)
stream.addFilter(console_filter)  # 🔍 Применяем фильтр только к консольному выводу


def get_colorful_logger(name: str = "main") -> logging.Logger:
    # Create and configure the logger
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)
    logger.addHandler(stream)
    logger.addFilter(basic_filter)  # 🔍 Базовый фильтр без подавления

    return logger


# Set up the root logger with the same formatting
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
root_logger.addHandler(stream)
root_logger.addFilter(basic_filter)  # 🔍 Базовый фильтр - позволяет Sentry получать все логи

ignore_logs = ["_trace", "httpx", "_client", "atrace", "aiohttp", "_client", "ariadne", "graphql"]
for lgr in ignore_logs:
    loggr = logging.getLogger(lgr)
    loggr.setLevel(logging.CRITICAL)  # Подавляем все кроме критических ошибок

# Дополнительно подавляем логи GraphQL и Ariadne для ошибок авторизации
graphql_logger = logging.getLogger("graphql")
graphql_logger.setLevel(logging.CRITICAL)  # Подавляем ERROR уровень для авторизации

ariadne_logger = logging.getLogger("ariadne")
ariadne_logger.setLevel(logging.CRITICAL)  # Подавляем ERROR уровень для авторизации


# Создаем специальный фильтр для Ariadne
class AriadneAuthFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        message = record.getMessage()
        # Полностью блокируем логи связанные с авторизацией в Ariadne
        return not any(
            phrase in message
            for phrase in [
                "Требуется авторизация",
                "AuthorizationError",
                "GraphQL request:",
                "decorated_function",
                "load_drafts",
                "GetUserDocuments",
                "GetDrafts",
                "Traceback",
                "await gather",
                "handle_field_error",
                "await_result",
                "During handling",
                "raise error",
            ]
        )


# Применяем фильтр к Ariadne и GraphQL логгерам
ariadne_logger.addFilter(AriadneAuthFilter())
graphql_logger.addFilter(AriadneAuthFilter())

# Дополнительно блокируем логи на уровне модулей
for logger_name in ["ariadne.log_error", "ariadne", "graphql.execution.execute", "graphql.execution", "graphql.error"]:
    extra_logger = logging.getLogger(logger_name)
    extra_logger.setLevel(logging.CRITICAL)
    extra_logger.addFilter(AriadneAuthFilter())
    extra_logger.disabled = True  # Полностью отключаем этот логгер


def custom_error_formatter(error: GraphQLError, debug: bool = False) -> dict[Any, Any]:
    """
    Кастомный форматтер ошибок для подавления трейсбеков у ожидаемых ошибок авторизации.
    🔍 Логирует AuthorizationError как обычные события, не как исключения
    """
    # Преобразуем в словарь для работы с ним
    formatted_error: dict[str, Any] = {
        "message": error.message,
        "locations": getattr(error.formatted, "locations", []),
        "path": getattr(error.formatted, "path", []),
        "extensions": getattr(error.formatted, "extensions", {}),
    }

    # Для ошибок авторизации не показываем трейсбек и НЕ логируем вообще
    # Проверяем по имени класса для избежания циклических импортов
    if (
        error.original_error
        and hasattr(error.original_error, "__class__")
        and error.original_error.__class__.__name__ == "AuthorizationError"
    ):
        # Убираем extensions.exception если есть
        if "extensions" in formatted_error and "exception" in formatted_error["extensions"]:
            del formatted_error["extensions"]["exception"]
        # НЕ логируем ошибки авторизации - они ожидаемы и нормальны
    # Для остальных ошибок используем стандартное логирование
    elif debug and error.original_error:
        root_logger.error(f"GraphQL error: {error.message}", exc_info=error.original_error)
    else:
        root_logger.warning(f"GraphQL error: {error.message}")

    return formatted_error