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:
    """🔍 Фильтр для консольного вывода - подавляет спам авторизации"""
    # 🚫 Блокируем urllib3 логи от Sentry/GlitchTip
    if record.name.startswith("urllib3") and "glitchtip.dscrs.site" in record.getMessage():
        return False

    # 🚫 Блокируем httpcore trace логи (спам от HTTP запросов)
    if record.name.startswith("httpcore") and "_trace" in record.name:
        return False

    # 🚫 Блокируем повторяющиеся debug логи из резолверов
    message = record.getMessage()
    if record.levelno == logging.DEBUG and any(
        spam_phrase in message
        for spam_phrase in [
            "connect_tcp.started",
            "connect_tcp.complete",
            "start_tls.started",
            "start_tls.complete",
            "send_request_headers",
            "send_request_body",
            "receive_response_headers",
            "receive_response_body",
            "response_closed",
            "close.started",
            "close.complete",
            "Adding main_topic subquery",
            "Adding topics subquery",
            "get_shouts_with_links] Starting with limit=",
            "Applied Python sorting by views_count",
            "Raw cached data size:",
            "Attempting to get cached data for key:",
            "Данные получены из кеша по ключу",
        ]
    ):
        return False

    # 🚫 Блокируем OAuth ошибки от ботов (GPTBot, crawlers)
    if (
        record.levelno == logging.ERROR
        and "Missing OAuth state parameter" in message
        and any(
            bot_phrase in message
            for bot_phrase in [
                "GPTBot",
                "gptbot",
                "crawler",
                "bot",
                "spider",
                "x-openai-host-hash",
            ]
        )
    ):
        return False

    # 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.error.graphql_error.GraphQLError: Требуется авторизация",
            ]
        ):
            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",
    "urllib3",
    "httpcore",
]
for lgr in ignore_logs:
    loggr = logging.getLogger(lgr)
    loggr.setLevel(logging.CRITICAL)  # Подавляем все кроме критических ошибок

# 🚫 Специально подавляем urllib3 логи от Sentry/GlitchTip
urllib3_logger = logging.getLogger("urllib3.connectionpool")
urllib3_logger.setLevel(logging.CRITICAL)
urllib3_logger.disabled = True  # Полностью отключаем urllib3 логи

# 🚫 Подавляем httpcore trace логи (спам от HTTP запросов)
httpcore_logger = logging.getLogger("httpcore._trace")
httpcore_logger.setLevel(logging.CRITICAL)
httpcore_logger.disabled = True  # Полностью отключаем httpcore trace логи

# Дополнительно подавляем логи 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"])


# Применяем фильтр к 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