import functools import time import traceback import inspect from typing import Any, Callable, Optional from ..core.database_logger import get_logger, LogLevel, LogCategory def log_execution(category: LogCategory = LogCategory.SYSTEM, log_args: bool = False, log_result: bool = False, log_performance: bool = True): """Decorador para logging automático de execução de funções""" def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs): logger = get_logger() # Informações da função module_name = func.__module__ function_name = func.__name__ # Obter número da linha try: line_number = inspect.getsourcelines(func)[1] except: line_number = 0 # Preparar metadados metadata = { 'function_signature': str(inspect.signature(func)) } if log_args: metadata['args'] = str(args) metadata['kwargs'] = str(kwargs) start_time = time.time() try: # Log de início da execução logger.log( level=LogLevel.DEBUG, category=category, message=f"Iniciando execução da função {function_name}", module=module_name, function=function_name, line_number=line_number, metadata=metadata ) # Executar função result = func(*args, **kwargs) execution_time = time.time() - start_time # Preparar metadados do resultado result_metadata = metadata.copy() if log_result: result_metadata['result'] = str(result)[:1000] # Limitar tamanho # Log de sucesso logger.log( level=LogLevel.INFO, category=category, message=f"Função {function_name} executada com sucesso", module=module_name, function=function_name, line_number=line_number, metadata=result_metadata, execution_time=execution_time ) # Log de performance se habilitado if log_performance: logger.log_performance_metric( metric_name=f"{module_name}.{function_name}_execution_time", metric_value=execution_time, unit="seconds", category=category.value, metadata={'function': function_name, 'module': module_name} ) return result except Exception as e: execution_time = time.time() - start_time # Log de erro logger.log( level=LogLevel.ERROR, category=category, message=f"Erro na execução da função {function_name}: {str(e)}", module=module_name, function=function_name, line_number=line_number, metadata=metadata, stack_trace=traceback.format_exc(), execution_time=execution_time ) raise return wrapper return decorator def log_api_call(endpoint: str = None): """Decorador específico para logging de chamadas de API""" def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs): logger = get_logger() module_name = func.__module__ function_name = func.__name__ try: line_number = inspect.getsourcelines(func)[1] except: line_number = 0 # Extrair informações da requisição se disponível request_info = {} if 'request' in kwargs: request = kwargs['request'] request_info = { 'method': getattr(request, 'method', 'UNKNOWN'), 'url': getattr(request, 'url', 'UNKNOWN'), 'user_agent': getattr(request, 'headers', {}).get('user-agent', 'UNKNOWN') } metadata = { 'endpoint': endpoint or function_name, 'request_info': request_info } start_time = time.time() try: # Log início da chamada API logger.log( level=LogLevel.INFO, category=LogCategory.API, message=f"Chamada API iniciada: {endpoint or function_name}", module=module_name, function=function_name, line_number=line_number, metadata=metadata ) result = func(*args, **kwargs) execution_time = time.time() - start_time # Log sucesso da API logger.log( level=LogLevel.INFO, category=LogCategory.API, message=f"Chamada API concluída com sucesso: {endpoint or function_name}", module=module_name, function=function_name, line_number=line_number, metadata=metadata, execution_time=execution_time ) # Métrica de performance da API logger.log_performance_metric( metric_name=f"api_{endpoint or function_name}_response_time", metric_value=execution_time, unit="seconds", category="API", metadata={'endpoint': endpoint or function_name} ) return result except Exception as e: execution_time = time.time() - start_time logger.log( level=LogLevel.ERROR, category=LogCategory.API, message=f"Erro na chamada API {endpoint or function_name}: {str(e)}", module=module_name, function=function_name, line_number=line_number, metadata=metadata, stack_trace=traceback.format_exc(), execution_time=execution_time ) raise return wrapper return decorator def log_ai_model_usage(model_name: str = None): """Decorador para logging de uso de modelos de IA""" def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs): logger = get_logger() module_name = func.__module__ function_name = func.__name__ try: line_number = inspect.getsourcelines(func)[1] except: line_number = 0 metadata = { 'model_name': model_name or function_name, 'input_size': len(str(args)) + len(str(kwargs)) } start_time = time.time() try: logger.log( level=LogLevel.INFO, category=LogCategory.AI_MODEL, message=f"Iniciando processamento do modelo {model_name or function_name}", module=module_name, function=function_name, line_number=line_number, metadata=metadata ) result = func(*args, **kwargs) execution_time = time.time() - start_time # Atualizar metadados com informações do resultado result_metadata = metadata.copy() result_metadata['output_size'] = len(str(result)) logger.log( level=LogLevel.INFO, category=LogCategory.AI_MODEL, message=f"Modelo {model_name or function_name} processado com sucesso", module=module_name, function=function_name, line_number=line_number, metadata=result_metadata, execution_time=execution_time ) # Métricas específicas de IA logger.log_performance_metric( metric_name=f"ai_model_{model_name or function_name}_inference_time", metric_value=execution_time, unit="seconds", category="AI_MODEL", metadata={ 'model': model_name or function_name, 'input_size': metadata['input_size'], 'output_size': result_metadata['output_size'] } ) return result except Exception as e: execution_time = time.time() - start_time logger.log( level=LogLevel.ERROR, category=LogCategory.AI_MODEL, message=f"Erro no modelo {model_name or function_name}: {str(e)}", module=module_name, function=function_name, line_number=line_number, metadata=metadata, stack_trace=traceback.format_exc(), execution_time=execution_time ) raise return wrapper return decorator class LoggingContext: """Context manager para logging de blocos de código""" def __init__(self, operation_name: str, category: LogCategory = LogCategory.SYSTEM, level: LogLevel = LogLevel.INFO, metadata: dict = None): self.operation_name = operation_name self.category = category self.level = level self.metadata = metadata or {} self.logger = get_logger() self.start_time = None def __enter__(self): self.start_time = time.time() # Obter informações do caller frame = inspect.currentframe().f_back module_name = frame.f_globals.get('__name__', 'unknown') function_name = frame.f_code.co_name line_number = frame.f_lineno self.logger.log( level=self.level, category=self.category, message=f"Iniciando operação: {self.operation_name}", module=module_name, function=function_name, line_number=line_number, metadata=self.metadata ) return self def __exit__(self, exc_type, exc_val, exc_tb): execution_time = time.time() - self.start_time # Obter informações do caller frame = inspect.currentframe().f_back module_name = frame.f_globals.get('__name__', 'unknown') function_name = frame.f_code.co_name line_number = frame.f_lineno if exc_type is None: # Sucesso self.logger.log( level=self.level, category=self.category, message=f"Operação concluída com sucesso: {self.operation_name}", module=module_name, function=function_name, line_number=line_number, metadata=self.metadata, execution_time=execution_time ) else: # Erro self.logger.log( level=LogLevel.ERROR, category=self.category, message=f"Erro na operação {self.operation_name}: {str(exc_val)}", module=module_name, function=function_name, line_number=line_number, metadata=self.metadata, stack_trace=traceback.format_exc(), execution_time=execution_time ) # Log de performance self.logger.log_performance_metric( metric_name=f"operation_{self.operation_name.replace(' ', '_')}_time", metric_value=execution_time, unit="seconds", category=self.category.value, metadata={'operation': self.operation_name} ) def quick_log(message: str, level: LogLevel = LogLevel.INFO, category: LogCategory = LogCategory.SYSTEM, **kwargs): """Função utilitária para logging rápido""" logger = get_logger() # Obter informações do caller frame = inspect.currentframe().f_back module_name = frame.f_globals.get('__name__', 'unknown') function_name = frame.f_code.co_name line_number = frame.f_lineno logger.log( level=level, category=category, message=message, module=module_name, function=function_name, line_number=line_number, **kwargs )