File size: 6,885 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | """
群聊相关异常
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from django.utils.functional import Promise
from .base import BusinessException
__all__: list[str] = [
"ChatProviderException",
"UnsupportedPlatformException",
"ChatCreationException",
"MessageSendException",
"ConfigurationException",
"OwnerSettingException",
# 快捷构造函数（替代原子类）
"owner_permission_error",
"owner_not_found_error",
"owner_validation_error",
"owner_retry_error",
"owner_timeout_error",
"owner_network_error",
"owner_config_error",
# 向后兼容别名
"OwnerPermissionException",
"OwnerNotFoundException",
"OwnerValidationException",
"OwnerRetryException",
"OwnerTimeoutException",
"OwnerNetworkException",
"OwnerConfigException",
]
class ChatProviderException(BusinessException):
"""群聊提供者异常基类"""
def __init__(
self,
message: str | Promise,
code: str | None = None,
errors: dict[str, Any] | None = None,
error_code: str | None = None,
platform: str | None = None,
) -> None:
super().__init__(message=message, code=code or "CHAT_PROVIDER_ERROR", errors=errors)
self.error_code = error_code
self.platform = platform
class UnsupportedPlatformException(ChatProviderException):
"""不支持的平台异常"""
def __init__(
self,
message: str | Promise = "不支持的群聊平台",
platform: str | None = None,
code: str | None = None,
errors: dict[str, Any] | None = None,
) -> None:
super().__init__(message=message, code=code or "UNSUPPORTED_PLATFORM", errors=errors, platform=platform)
class ChatCreationException(ChatProviderException):
"""群聊创建失败异常"""
def __init__(
self,
message: str | Promise = "群聊创建失败",
code: str | None = None,
errors: dict[str, Any] | None = None,
error_code: str | None = None,
platform: str | None = None,
) -> None:
super().__init__(
message=message, code=code or "CHAT_CREATION_ERROR", errors=errors, error_code=error_code, platform=platform
)
class MessageSendException(ChatProviderException):
"""消息发送失败异常"""
def __init__(
self,
message: str | Promise = "消息发送失败",
code: str | None = None,
errors: dict[str, Any] | None = None,
error_code: str | None = None,
platform: str | None = None,
chat_id: str | None = None,
) -> None:
super().__init__(
message=message, code=code or "MESSAGE_SEND_ERROR", errors=errors, error_code=error_code, platform=platform
)
self.chat_id = chat_id
class ConfigurationException(ChatProviderException):
"""配置错误异常"""
def __init__(
self,
message: str | Promise = "群聊平台配置错误",
code: str | None = None,
errors: dict[str, Any] | None = None,
platform: str | None = None,
missing_config: str | None = None,
) -> None:
super().__init__(message=message, code=code or "CONFIGURATION_ERROR", errors=errors, platform=platform)
self.missing_config = missing_config
class OwnerSettingException(ChatProviderException):
"""
群主设置异常（统一基类）
通过 code 字段区分具体错误类型：
- OWNER_PERMISSION_ERROR: 群主权限不足
- OWNER_NOT_FOUND: 群主用户不存在
- OWNER_VALIDATION_ERROR: 群主验证失败
- OWNER_RETRY_ERROR: 群主设置重试失败
- OWNER_TIMEOUT_ERROR: 群主设置操作超时
- OWNER_NETWORK_ERROR: 群主设置网络错误
- OWNER_CONFIG_ERROR: 群主配置错误
"""
def __init__(
self,
message: str | Promise,
code: str | None = None,
errors: dict[str, Any] | None = None,
error_code: str | None = None,
platform: str | None = None,
owner_id: str | None = None,
chat_id: str | None = None,
**extra: Any,
) -> None:
super().__init__(
message=message, code=code or "OWNER_SETTING_ERROR", errors=errors, error_code=error_code, platform=platform
)
self.owner_id = owner_id
self.chat_id = chat_id
for key, value in extra.items():
setattr(self, key, value)
# ── 快捷构造函数（替代原子类）──────────────────────────────────────────────────
def owner_permission_error(message: str | Promise = "群主权限不足", **kwargs: Any) -> OwnerSettingException:
"""群主权限不足"""
return OwnerSettingException(message=message, code="OWNER_PERMISSION_ERROR", **kwargs)
def owner_not_found_error(message: str | Promise = "群主用户不存在", **kwargs: Any) -> OwnerSettingException:
"""群主用户不存在"""
return OwnerSettingException(message=message, code="OWNER_NOT_FOUND", **kwargs)
def owner_validation_error(message: str | Promise = "群主验证失败", **kwargs: Any) -> OwnerSettingException:
"""群主验证失败"""
return OwnerSettingException(message=message, code="OWNER_VALIDATION_ERROR", **kwargs)
def owner_retry_error(message: str | Promise = "群主设置重试失败", **kwargs: Any) -> OwnerSettingException:
"""群主设置重试失败"""
return OwnerSettingException(message=message, code="OWNER_RETRY_ERROR", **kwargs)
def owner_timeout_error(message: str | Promise = "群主设置操作超时", **kwargs: Any) -> OwnerSettingException:
"""群主设置操作超时"""
return OwnerSettingException(message=message, code="OWNER_TIMEOUT_ERROR", **kwargs)
def owner_network_error(message: str | Promise = "群主设置网络错误", **kwargs: Any) -> OwnerSettingException:
"""群主设置网络错误"""
return OwnerSettingException(message=message, code="OWNER_NETWORK_ERROR", **kwargs)
def owner_config_error(message: str | Promise = "群主配置错误", **kwargs: Any) -> OwnerSettingException:
"""群主配置错误"""
return OwnerSettingException(message=message, code="OWNER_CONFIG_ERROR", **kwargs)
# ── 向后兼容别名（保持 isinstance 检查可用）────────────────────────────────────
OwnerPermissionException = OwnerSettingException
OwnerNotFoundException = OwnerSettingException
OwnerValidationException = OwnerSettingException
OwnerRetryException = OwnerSettingException
OwnerTimeoutException = OwnerSettingException
OwnerNetworkException = OwnerSettingException
OwnerConfigException = OwnerSettingException
|