Spaces:
Sleeping
Sleeping
| 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: | |
| 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: | |
| 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: | |
| 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 | |
| ) |