"""Módulo de logging para requisições e respostas HTTP.""" import json import time from datetime import datetime from typing import Dict, Any, Optional from functools import wraps import inspect class RequestResponseLogger: """Logger para capturar dados de requisições e respostas.""" def __init__(self, enabled: bool = True): """Inicializa o logger. Args: enabled: Se o logging está habilitado """ self.enabled = enabled self.request_counter = 0 def log_request(self, endpoint: str, data: Dict[str, Any], method: str = "POST") -> str: """Registra dados da requisição recebida. Args: endpoint: Nome do endpoint/função chamada data: Dados recebidos na requisição method: Método HTTP (GET, POST, etc.) Returns: ID único da requisição """ if not self.enabled: return "" self.request_counter += 1 request_id = f"REQ_{self.request_counter}_{int(time.time())}" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] print("\n" + "="*80) print(f"📥 REQUISIÇÃO RECEBIDA [{request_id}]") print(f"⏰ Timestamp: {timestamp}") print(f"🎯 Endpoint: {endpoint}") print(f"🔧 Método: {method}") print("📋 Dados Recebidos:") # Formatar dados de forma legível try: formatted_data = json.dumps(data, indent=2, ensure_ascii=False, default=str) print(formatted_data) except Exception as e: print(f" {data} (Erro na formatação JSON: {e})") print("="*80) return request_id def log_response(self, request_id: str, response_data: Any, processing_time: float = None) -> None: """Registra dados da resposta enviada. Args: request_id: ID da requisição correspondente response_data: Dados enviados como resposta processing_time: Tempo de processamento em segundos """ if not self.enabled: return timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] print("\n" + "-"*80) print(f"📤 RESPOSTA ENVIADA [{request_id}]") print(f"⏰ Timestamp: {timestamp}") if processing_time is not None: print(f"⚡ Tempo de Processamento: {processing_time:.3f}s") print("📋 Dados Enviados:") # Formatar resposta de forma legível try: if isinstance(response_data, (dict, list)): formatted_response = json.dumps(response_data, indent=2, ensure_ascii=False, default=str) print(formatted_response) elif isinstance(response_data, str) and len(response_data) > 500: # Para strings muito longas, mostrar apenas o início e fim print(f" [STRING LONGA - {len(response_data)} caracteres]") print(f" Início: {response_data[:200]}...") print(f" Fim: ...{response_data[-200:]}") else: print(f" {response_data}") except Exception as e: print(f" {response_data} (Erro na formatação: {e})") print("-"*80 + "\n") def log_error(self, request_id: str, error: Exception, processing_time: float = None) -> None: """Registra erro durante o processamento. Args: request_id: ID da requisição correspondente error: Exceção ocorrida processing_time: Tempo até o erro em segundos """ if not self.enabled: return timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] print("\n" + "!"*80) print(f"❌ ERRO NO PROCESSAMENTO [{request_id}]") print(f"⏰ Timestamp: {timestamp}") if processing_time is not None: print(f"⚡ Tempo até Erro: {processing_time:.3f}s") print(f"🚨 Tipo do Erro: {type(error).__name__}") print(f"📝 Mensagem: {str(error)}") print("!"*80 + "\n") # Instância global do logger request_logger = RequestResponseLogger() def log_requests_responses(endpoint_name: str = None): """Decorator para automaticamente logar requisições e respostas de funções. Args: endpoint_name: Nome personalizado para o endpoint (opcional) """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): if not request_logger.enabled: return func(*args, **kwargs) # Determinar nome do endpoint endpoint = endpoint_name or func.__name__ # Capturar dados da requisição request_data = { 'args': args, 'kwargs': kwargs, 'function_signature': str(inspect.signature(func)) } # Logar requisição request_id = request_logger.log_request(endpoint, request_data) # Executar função e medir tempo start_time = time.time() try: result = func(*args, **kwargs) processing_time = time.time() - start_time # Logar resposta request_logger.log_response(request_id, result, processing_time) return result except Exception as e: processing_time = time.time() - start_time # Logar erro request_logger.log_error(request_id, e, processing_time) # Re-raise a exceção raise return wrapper return decorator def enable_logging(): """Habilita o logging de requisições/respostas.""" request_logger.enabled = True print("✅ Logging de requisições/respostas HABILITADO") def disable_logging(): """Desabilita o logging de requisições/respostas.""" request_logger.enabled = False print("❌ Logging de requisições/respostas DESABILITADO") def get_logging_status() -> bool: """Retorna o status atual do logging. Returns: True se o logging está habilitado, False caso contrário """ return request_logger.enabled