121 lines
3.6 KiB
Python
121 lines
3.6 KiB
Python
|
"""UI 自动化错误码分类与异常体系。
|
|||
|
|
|
|||
|
|
定义可重试/不可重试错误码集合，以及 Flow 异常体系。
|
|||
|
|
供 RetryPolicy.should_retry / CircuitBreaker / Flow 状态机使用。
|
|||
|
|
|
|||
|
|
错误码分类（spec 第十节）：
|
|||
|
|
- RETRYABLE_CODES（4 类）：瞬时故障，最多重试 1 次
|
|||
|
|
- NON_RETRYABLE_CODES（7 类）：永久故障，不重试
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from typing import Optional
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 可重试错误码：瞬时故障，RetryPolicy.should_retry 返回 True
|
|||
|
|
RETRYABLE_CODES: frozenset[str] = frozenset({
|
|||
|
|
"RATE_LIMITED", # 限流，需退避后重试
|
|||
|
|
"SEND_TIMEOUT", # 发送超时，可能下次成功
|
|||
|
|
"STATE_DIRTY", # UI 状态脏，重置后可重试
|
|||
|
|
"X11_TEMP_UNAVAILABLE", # X11 临时不可用，重试可能恢复
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
# 不可重试错误码：永久故障，重试无意义
|
|||
|
|
NON_RETRYABLE_CODES: frozenset[str] = frozenset({
|
|||
|
|
"WECHAT_NOT_LOGGED_IN", # 未登录，需人工介入
|
|||
|
|
"WINDOW_NOT_FOUND", # 窗口未找到，需检查微信运行
|
|||
|
|
"ELEMENT_NOT_FOUND", # 元素未找到，需检查 UI 布局
|
|||
|
|
"DB_VERIFY_FAILED", # DB 校验失败（防串号），需人工排查
|
|||
|
|
"AMBIGUOUS_CONTACT", # 联系人歧义，需修正 display_name
|
|||
|
|
"X11_UNAVAILABLE", # X11 持续不可用，需检查容器
|
|||
|
|
"BRIDGE_CIRCUITED", # bridge 熔断，需等待恢复
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
|
|||
|
|
def is_retryable(code: str) -> bool:
|
|||
|
|
"""判断错误码是否可重试。"""
|
|||
|
|
return code in RETRYABLE_CODES
|
|||
|
|
|
|||
|
|
|
|||
|
|
class FlowError(Exception):
|
|||
|
|
"""Flow 执行异常基类。
|
|||
|
|
|
|||
|
|
Attributes:
|
|||
|
|
code: 错误码（如 WINDOW_NOT_FOUND / SEND_TIMEOUT）
|
|||
|
|
message: 人类可读错误信息
|
|||
|
|
retryable: 是否可重试（根据 code 自动推断，可显式覆盖）
|
|||
|
|
details: 额外上下文（dict，可选）
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(
|
|||
|
|
self,
|
|||
|
|
code: str,
|
|||
|
|
message: str = "",
|
|||
|
|
*,
|
|||
|
|
retryable: Optional[bool] = None,
|
|||
|
|
details: Optional[dict] = None,
|
|||
|
|
) -> None:
|
|||
|
|
self.code = code
|
|||
|
|
self.message = message or code
|
|||
|
|
# 未显式指定时按 code 推断
|
|||
|
|
self.retryable = (
|
|||
|
|
retryable if retryable is not None else is_retryable(code)
|
|||
|
|
)
|
|||
|
|
self.details = details or {}
|
|||
|
|
super().__init__(self.message)
|
|||
|
|
|
|||
|
|
def __repr__(self) -> str:
|
|||
|
|
return (
|
|||
|
|
f"FlowError(code={self.code!r}, retryable={self.retryable}, "
|
|||
|
|
f"message={self.message!r})"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TransientError(FlowError):
|
|||
|
|
"""瞬时错误（可重试）。
|
|||
|
|
|
|||
|
|
用于 RETRYABLE_CODES 中的错误码，或显式指定 retryable=True。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(
|
|||
|
|
self,
|
|||
|
|
code: str,
|
|||
|
|
message: str = "",
|
|||
|
|
*,
|
|||
|
|
details: Optional[dict] = None,
|
|||
|
|
) -> None:
|
|||
|
|
super().__init__(code, message, retryable=True, details=details)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class PermanentError(FlowError):
|
|||
|
|
"""永久错误（不可重试）。
|
|||
|
|
|
|||
|
|
用于 NON_RETRYABLE_CODES 中的错误码，或显式指定 retryable=False。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(
|
|||
|
|
self,
|
|||
|
|
code: str,
|
|||
|
|
message: str = "",
|
|||
|
|
*,
|
|||
|
|
details: Optional[dict] = None,
|
|||
|
|
) -> None:
|
|||
|
|
super().__init__(code, message, retryable=False, details=details)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class ElementNotFoundError(FlowError):
|
|||
|
|
"""元素未找到（图像匹配失败且 require_image=True，或几何规格缺失）。
|
|||
|
|
|
|||
|
|
默认 code=ELEMENT_NOT_FOUND，retryable=False。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(
|
|||
|
|
self,
|
|||
|
|
message: str = "",
|
|||
|
|
*,
|
|||
|
|
code: str = "ELEMENT_NOT_FOUND",
|
|||
|
|
details: Optional[dict] = None,
|
|||
|
|
) -> None:
|
|||
|
|
super().__init__(code, message, retryable=False, details=details)
|