170 lines
6.5 KiB
Python
170 lines
6.5 KiB
Python
"""
|
||
自定义异常类定义
|
||
"""
|
||
|
||
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 |