""" 自定义异常类定义 """ from typing import Optional, Dict, Any class SocialMediaError(Exception): """基础异常类""" def __init__(self, message: str, error_code: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message) self.message = message self.error_code = error_code self.details = details or {} def to_dict(self) -> Dict[str, Any]: """转换为字典格式""" return { "error_type": self.__class__.__name__, "message": self.message, "error_code": self.error_code, "details": self.details } class LoginFailedError(SocialMediaError): """登录失败异常""" def __init__(self, message: str = "登录失败", platform: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "LOGIN_FAILED", details) self.platform = platform class UploadFailedError(SocialMediaError): """上传失败异常""" def __init__(self, message: str = "文件上传失败", file_path: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "UPLOAD_FAILED", details) self.file_path = file_path class ContentRejectedError(SocialMediaError): """内容被拒绝异常""" def __init__(self, message: str = "内容被平台拒绝", reason: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "CONTENT_REJECTED", details) self.reason = reason class RateLimitError(SocialMediaError): """频率限制异常""" def __init__(self, message: str = "操作频率过高,请稍后再试", retry_after: Optional[int] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "RATE_LIMIT", details) self.retry_after = retry_after class ElementNotFoundError(SocialMediaError): """元素未找到异常""" def __init__(self, message: str = "页面元素未找到", selector: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "ELEMENT_NOT_FOUND", details) self.selector = selector class TimeoutError(SocialMediaError): """超时异常""" def __init__(self, message: str = "操作超时", timeout: Optional[float] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "TIMEOUT", details) self.timeout = timeout class AuthenticationError(SocialMediaError): """认证失败异常""" def __init__(self, message: str = "认证失败", platform: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "AUTHENTICATION_FAILED", details) self.platform = platform class NetworkError(SocialMediaError): """网络错误异常""" def __init__(self, message: str = "网络连接错误", url: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "NETWORK_ERROR", details) self.url = url class ValidationError(SocialMediaError): """数据验证错误异常""" def __init__(self, message: str = "数据验证失败", field: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "VALIDATION_ERROR", details) self.field = field class PlatformNotSupportedError(SocialMediaError): """平台不支持异常""" def __init__(self, message: str = "不支持的平台", platform: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "PLATFORM_NOT_SUPPORTED", details) self.platform = platform class BrowserError(SocialMediaError): """浏览器错误异常""" def __init__(self, message: str = "浏览器操作错误", details: Optional[Dict[str, Any]] = None): super().__init__(message, "BROWSER_ERROR", details) class CookieExpiredError(AuthenticationError): """Cookie过期异常""" def __init__(self, message: str = "Cookie已过期,请重新登录", platform: Optional[str] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, platform, details) self.error_code = "COOKIE_EXPIRED" class FileSizeExceededError(ValidationError): """文件大小超限异常""" def __init__(self, message: str = "文件大小超出限制", file_size: Optional[int] = None, max_size: Optional[int] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "FILE_SIZE_EXCEEDED", details) self.file_size = file_size self.max_size = max_size class UnsupportedFormatError(ValidationError): """不支持的文件格式异常""" def __init__(self, message: str = "不支持的文件格式", file_format: Optional[str] = None, supported_formats: Optional[list] = None, details: Optional[Dict[str, Any]] = None): super().__init__(message, "UNSUPPORTED_FORMAT", details) self.file_format = file_format self.supported_formats = supported_formats # 异常处理装饰器 def handle_exceptions(platform: Optional[str] = None, default_message: str = "操作失败"): """异常处理装饰器""" def decorator(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except SocialMediaError: raise # 重新抛出自定义异常 except Exception as e: # 将未知异常包装为SocialMediaError raise SocialMediaError( message=f"{default_message}: {str(e)}", error_code="UNKNOWN_ERROR", details={ "platform": platform, "original_exception": str(e), "function": func.__name__ } ) return wrapper return decorator # 重试机制 class RetryStrategy: """重试策略类""" def __init__(self, max_retries: int = 3, backoff_factor: float = 2.0): self.max_retries = max_retries self.backoff_factor = backoff_factor self.retryable_exceptions = ( NetworkError, TimeoutError, ElementNotFoundError, UploadFailedError ) def should_retry(self, exception: Exception, attempt: int) -> bool: """判断是否应该重试""" return attempt < self.max_retries and isinstance(exception, self.retryable_exceptions) def get_delay(self, attempt: int) -> float: """获取重试延迟时间""" return self.backoff_factor ** attempt