zhaoyuan_DataPreProcess/core/error_handler.py

import sys
import traceback
from typing import Callable, Any, Optional, Dict, List, Type, Union
from functools import wraps
import time

class ProcessingError(Exception):
    """数据处理错误的基类"""
    def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
        self.message = message
        self.details = details or {}
        super().__init__(message)

class DataValidationError(ProcessingError):
    """数据验证错误"""
    pass

class ResourceNotFoundError(ProcessingError):
    """资源未找到错误"""
    pass

class PluginError(ProcessingError):
    """插件相关错误"""
    pass

class ConfigurationError(ProcessingError):
    """配置错误"""
    pass

class EngineExecutionError(ProcessingError):
    """引擎执行错误"""
    pass

class ErrorHandler:
    """错误处理器，提供统一的异常处理机制"""
    
    @staticmethod
    def handle_exceptions(func: Callable) -> Callable:
        """装饰器：用于处理函数执行期间的异常"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                ErrorHandler.handle_exception(e, {"function": func.__name__})
                raise
        return wrapper

    @staticmethod
    def handle_exception(exc: Exception, context: Optional[Dict[str, Any]] = None) -> None:
        """处理异常，记录日志并提供上下文信息"""
        context = context or {}
        
        # 获取异常的堆栈跟踪
        exc_type, exc_value, exc_traceback = sys.exc_info()
        stack_trace = traceback.format_exception(exc_type, exc_value, exc_traceback)
        
        # 打印异常信息
        error_message = f"异常: {type(exc).__name__}: {str(exc)}"
        print(error_message)
        
        # 打印上下文信息
        if context:
            print(f"上下文信息: {context}")
        
        # 打印堆栈跟踪
        print("堆栈跟踪:\n" + "".join(stack_trace))
    
    @staticmethod
    def retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0, 
              exceptions: Union[Type[Exception], List[Type[Exception]]] = Exception):
        """重试装饰器，用于自动重试可能失败的操作
        
        Args:
            max_attempts: 最大尝试次数
            delay: 初始延迟时间（秒）
            backoff: 延迟时间的增长因子
            exceptions: 需要捕获的异常类型
        """
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                attempt = 1
                current_delay = delay
                
                while attempt <= max_attempts:
                    try:
                        return func(*args, **kwargs)
                    except exceptions as e:
                        if attempt == max_attempts:
                            print(f"重试{max_attempts}次后失败: {func.__name__}")
                            raise
                        
                        print(f"尝试 {attempt}/{max_attempts} 失败: {func.__name__}, 错误: {e}")
                        print(f"将在 {current_delay:.2f} 秒后重试...")
                        
                        time.sleep(current_delay)
                        attempt += 1
                        current_delay *= backoff
            
            return wrapper
        return decorator
    
    @staticmethod
    def validate_input(validator: Callable[[Any], bool], error_message: str = "输入验证失败"):
        """输入验证装饰器，用于验证函数输入参数
        
        Args:
            validator: 验证函数，接收函数的所有参数并返回布尔值
            error_message: 验证失败时的错误消息
        """
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                if not validator(*args, **kwargs):
                    raise DataValidationError(error_message)
                return func(*args, **kwargs)
            return wrapper
        return decorator
    
    @staticmethod
    def measure_performance(func):
        """性能测量装饰器，用于记录函数执行时间"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            result = func(*args, **kwargs)
            end_time = time.time()
            execution_time = end_time - start_time
            print(f"函数 {func.__name__} 执行时间: {execution_time:.4f} 秒")
            return result
        return wrapper
    
    @staticmethod
    def safe_execute(fallback_value: Any = None, log_exception: bool = True):
        """安全执行装饰器，捕获异常并返回默认值
        
        Args:
            fallback_value: 发生异常时返回的默认值
            log_exception: 是否记录异常信息
        """
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if log_exception:
                        print(f"函数 {func.__name__} 执行失败: {e}")
                    return fallback_value
            return wrapper
        return decorator