324 lines
12 KiB
Python
324 lines
12 KiB
Python
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
|
||
|
||
# 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",
|
||
"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",
|
||
"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
|