#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 统一异常处理模块 定义项目专用的异常类型和标准化的重试机制 """ import time import logging from typing import Optional, Callable, Any, Union from functools import wraps import os logger = logging.getLogger(__name__) class TravelContentCreatorError(Exception): """项目基础异常类""" def __init__(self, message: str, error_code: Optional[str] = None, details: Optional[dict] = None): super().__init__(message) self.message = message self.error_code = error_code self.details = details or {} def __str__(self): if self.error_code: return f"[{self.error_code}] {self.message}" return self.message class ConfigurationError(TravelContentCreatorError): """配置相关异常""" pass class AIModelError(TravelContentCreatorError): """AI模型调用异常""" pass class ContentGenerationError(TravelContentCreatorError): """内容生成异常""" pass class PosterGenerationError(TravelContentCreatorError): """海报生成异常""" pass class ResourceError(TravelContentCreatorError): """资源文件异常""" pass class ValidationError(TravelContentCreatorError): """数据验证异常""" pass class TimeoutError(TravelContentCreatorError): """超时异常""" pass class RetryableError(TravelContentCreatorError): """可重试异常""" pass class NonRetryableError(TravelContentCreatorError): """不可重试异常""" pass def retry_on_failure( max_retries: int = 3, delay: float = 1.0, backoff: float = 2.0, exceptions: tuple = (Exception,), non_retryable_exceptions: tuple = (NonRetryableError,) ): """ 重试装饰器 Args: max_retries: 最大重试次数 delay: 初始延迟时间（秒） backoff: 退避系数 exceptions: 需要重试的异常类型 non_retryable_exceptions: 不可重试的异常类型 """ def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs) -> Any: last_exception = None current_delay = delay for attempt in range(max_retries + 1): try: return func(*args, **kwargs) except non_retryable_exceptions as e: logger.error(f"不可重试异常: {e}") raise except exceptions as e: last_exception = e if attempt < max_retries: logger.warning(f"函数 {func.__name__} 第 {attempt + 1} 次调用失败: {e}") logger.info(f"将在 {current_delay:.1f} 秒后重试...") time.sleep(current_delay) current_delay *= backoff else: logger.error(f"函数 {func.__name__} 重试 {max_retries} 次后仍然失败") break # 如果所有重试都失败了，抛出最后一个异常 if last_exception: raise last_exception return wrapper return decorator def timeout_handler(timeout_seconds: int): """ 超时处理装饰器 注意: 这个装饰器使用 signal 模块，因此在非Unix-like系统（如Windows）上可能无法正常工作。 Args: timeout_seconds: 超时时间（秒） """ def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs) -> Any: # 在Windows上，signal模块功能有限，以下代码可能无法按预期工作 if os.name == 'nt': logger.warning("timeout_handler在Windows上功能受限，可能无法生效") return func(*args, **kwargs) import signal def timeout_signal_handler(signum, frame): raise TimeoutError(f"函数 {func.__name__} 执行超时 ({timeout_seconds}秒)") # 设置超时信号 old_handler = signal.signal(signal.SIGALRM, timeout_signal_handler) signal.alarm(timeout_seconds) try: result = func(*args, **kwargs) return result finally: # 取消超时信号 signal.alarm(0) signal.signal(signal.SIGALRM, old_handler) return wrapper return decorator def handle_exceptions( default_return: Any = None, log_level: str = "ERROR", reraise: bool = True ): """ 通用异常处理装饰器 Args: default_return: 异常时的默认返回值 log_level: 日志级别 reraise: 是否重新抛出异常 """ def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs) -> Any: try: return func(*args, **kwargs) except Exception as e: # 记录异常 log_func = getattr(logger, log_level.lower()) log_func(f"函数 {func.__name__} 发生异常: {e}", exc_info=True) if reraise: raise else: return default_return return wrapper return decorator class ErrorContext: """错误上下文管理器""" def __init__(self, operation: str, error_type: type = TravelContentCreatorError): self.operation = operation self.error_type = error_type self.start_time = None def __enter__(self): self.start_time = time.time() logger.info(f"开始执行操作: {self.operation}") return self def __exit__(self, exc_type, exc_val, exc_tb): duration = time.time() - self.start_time if exc_type is None: logger.info(f"操作完成: {self.operation} (耗时: {duration:.2f}秒)") else: logger.error(f"操作失败: {self.operation} (耗时: {duration:.2f}秒)") # 如果是预期的异常类型，包装后重新抛出 if not isinstance(exc_val, TravelContentCreatorError): wrapped_error = self.error_type( f"操作 '{self.operation}' 失败: {str(exc_val)}", details={'original_error': str(exc_val), 'duration': duration} ) raise wrapped_error from exc_val return False def validate_required_fields(data: dict, required_fields: list, error_prefix: str = ""): """ 验证必需字段 Args: data: 要验证的数据字典 required_fields: 必需字段列表 error_prefix: 错误消息前缀 Raises: ValidationError: 当必需字段缺失时 """ missing_fields = [] for field in required_fields: if field not in data or data[field] is None: missing_fields.append(field) if missing_fields: error_msg = f"{error_prefix}缺少必需字段: {', '.join(missing_fields)}" raise ValidationError(error_msg, error_code="MISSING_REQUIRED_FIELDS") def safe_execute(func: Callable, *args, **kwargs) -> tuple: """ 安全执行函数，返回 (success, result_or_error) Args: func: 要执行的函数 *args: 函数参数 **kwargs: 函数关键字参数 Returns: tuple: (是否成功, 结果或异常对象) """ try: result = func(*args, **kwargs) return True, result except Exception as e: logger.exception(f"函数 {func.__name__} 执行失败") return False, e # 常用的异常实例 COMMON_ERRORS = { 'config_missing': ConfigurationError("配置文件缺失或格式错误"), 'api_key_missing': ConfigurationError("API密钥未配置"), 'model_unavailable': AIModelError("AI模型不可用"), 'network_error': RetryableError("网络连接错误"), 'resource_not_found': ResourceError("资源文件未找到"), 'invalid_format': ValidationError("数据格式无效"), } def get_error_by_code(error_code: str) -> TravelContentCreatorError: """根据错误代码获取预定义的异常对象""" return COMMON_ERRORS.get(error_code, TravelContentCreatorError(f"未知错误: {error_code}"))