""" Декораторы для автоматического логирования функций """ import asyncio import inspect from functools import wraps from typing import Callable, Any, Optional, Dict, Union from aiogram.types import Message, CallbackQuery from services.infrastructure.logger import get_logger def log_function_call( function_name: Optional[str] = None, log_params: bool = True, log_result: bool = False, log_level: str = "info", quiet: bool = False ): """ Декоратор для автоматического логирования входа/выхода из функций Args: function_name: Кастомное имя функции для логов (по умолчанию берется из func.__name__) log_params: Логировать ли параметры вызова log_result: Логировать ли результат выполнения log_level: Уровень логирования ('info', 'debug', 'warning') quiet: Тихое логирование (только ошибки) """ def decorator(func: Callable) -> Callable: @wraps(func) async def async_wrapper(*args, **kwargs): logger = get_logger(func.__module__) name = function_name or func.__name__ # Формируем контекстную информацию context_info = _build_context_info(args, kwargs, log_params) # Логируем вход в функцию (только если не тихий режим) if not quiet: log_method = getattr(logger, log_level) log_method(f"🚀 Начало выполнения {name}{context_info}") try: result = await func(*args, **kwargs) # Логируем успешное завершение (только если не тихий режим) if not quiet: result_info = "" if log_result and result is not None: result_info = f" | Результат: {_format_result(result)}" log_method(f"✅ Успешное завершение {name}{result_info}") return result except Exception as e: # Логируем ошибку (всегда, даже в тихом режиме) logger.error(f"❌ Ошибка в {name}: {e}") raise @wraps(func) def sync_wrapper(*args, **kwargs): logger = get_logger(func.__module__) name = function_name or func.__name__ # Формируем контекстную информацию context_info = _build_context_info(args, kwargs, log_params) # Логируем вход в функцию (только если не тихий режим) if not quiet: log_method = getattr(logger, log_level) log_method(f"🚀 Начало выполнения {name}{context_info}") try: result = func(*args, **kwargs) # Логируем успешное завершение (только если не тихий режим) if not quiet: result_info = "" if log_result and result is not None: result_info = f" | Результат: {_format_result(result)}" log_method(f"✅ Успешное завершение {name}{result_info}") return result except Exception as e: # Логируем ошибку (всегда, даже в тихом режиме) logger.error(f"❌ Ошибка в {name}: {e}") raise # Возвращаем правильный wrapper в зависимости от типа функции return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper return decorator def log_business_event( event_name: str, log_params: bool = True, log_result: bool = True ): """ Декоратор для логирования бизнес-событий Args: event_name: Название бизнес-события log_params: Логировать ли параметры log_result: Логировать ли результат """ def decorator(func: Callable) -> Callable: @wraps(func) async def async_wrapper(*args, **kwargs): logger = get_logger(func.__module__) # Формируем контекстную информацию context_info = _build_context_info(args, kwargs, log_params) # Логируем бизнес-событие logger.info(f"📊 Бизнес-событие: {event_name}{context_info}") try: result = await func(*args, **kwargs) # Логируем результат бизнес-события if log_result and result is not None: result_info = _format_result(result) logger.info(f"📈 Результат {event_name}: {result_info}") return result except Exception as e: logger.error(f"💥 Ошибка в бизнес-событии {event_name}: {e}") raise @wraps(func) def sync_wrapper(*args, **kwargs): logger = get_logger(func.__module__) # Формируем контекстную информацию context_info = _build_context_info(args, kwargs, log_params) # Логируем бизнес-событие logger.info(f"📊 Бизнес-событие: {event_name}{context_info}") try: result = func(*args, **kwargs) # Логируем результат бизнес-события if log_result and result is not None: result_info = _format_result(result) logger.info(f"📈 Результат {event_name}: {result_info}") return result except Exception as e: logger.error(f"💥 Ошибка в бизнес-событии {event_name}: {e}") raise return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper return decorator def log_fsm_transition( from_state: Optional[str] = None, to_state: Optional[str] = None ): """ Декоратор для логирования переходов FSM состояний Args: from_state: Исходное состояние (если None, будет определено автоматически) to_state: Целевое состояние (если None, будет определено автоматически) """ def decorator(func: Callable) -> Callable: @wraps(func) async def async_wrapper(*args, **kwargs): logger = get_logger(func.__module__) # Извлекаем FSM context из аргументов fsm_context = None for arg in args: if hasattr(arg, 'get_state') and hasattr(arg, 'set_state'): fsm_context = arg break # Логируем переход состояния if fsm_context: current_state = await fsm_context.get_state() logger.info(f"🔄 FSM переход: {current_state} -> {to_state or 'новое состояние'}") try: result = await func(*args, **kwargs) # Логируем успешный переход if fsm_context: new_state = await fsm_context.get_state() logger.info(f"✅ FSM переход завершен: {from_state or 'предыдущее состояние'} -> {new_state}") return result except Exception as e: logger.error(f"❌ Ошибка в FSM переходе: {e}") raise return async_wrapper return decorator def _build_context_info(args: tuple, kwargs: dict, log_params: bool) -> str: """Построение контекстной информации для логов""" if not log_params: return "" context_parts = [] # Извлекаем информацию о пользователе из аргументов user_id = None for arg in args: if isinstance(arg, (Message, CallbackQuery)): user_id = arg.from_user.id context_parts.append(f"user_id={user_id}") break elif hasattr(arg, 'from_user_id'): user_id = arg.from_user_id context_parts.append(f"from_user_id={user_id}") elif hasattr(arg, 'to_user_id'): context_parts.append(f"to_user_id={arg.to_user_id}") elif hasattr(arg, 'id') and isinstance(arg.id, int): context_parts.append(f"id={arg.id}") # Добавляем важные параметры из kwargs important_params = ['question_id', 'user_id', 'page', 'limit', 'status'] for param in important_params: if param in kwargs and kwargs[param] is not None: context_parts.append(f"{param}={kwargs[param]}") return f" | {', '.join(context_parts)}" if context_parts else "" def _format_result(result: Any) -> str: """Форматирование результата для логов""" if result is None: return "None" if isinstance(result, (str, int, float, bool)): return str(result) if hasattr(result, 'id'): return f"id={result.id}" if isinstance(result, (list, tuple)): return f"count={len(result)}" if isinstance(result, dict): return f"keys={list(result.keys())}" return str(type(result).__name__) # Удобные алиасы для часто используемых декораторов log_handler = log_function_call log_service = log_function_call log_business = log_business_event log_fsm = log_fsm_transition # Тихие декораторы для middleware и служебных функций log_quiet = lambda **kwargs: log_function_call(quiet=True, **kwargs) log_middleware = lambda **kwargs: log_function_call(quiet=True, log_level="debug", **kwargs) # Декоратор для служебных функций (только ошибки) def log_utility(func: Callable) -> Callable: """Декоратор для служебных функций - логирует только ошибки""" return log_function_call(quiet=True, log_params=False, log_result=False)(func)