""" 错误处理和重试机制 """ import logging import time from typing import Callable, Any, Dict, Optional from functools import wraps logger = logging.getLogger(__name__) def retry(max_retries: int = 3, delay: float = 1.0, backoff: float = 2.0, exceptions: tuple = (Exception,)): """ 重试装饰器 Args: max_retries: 最大重试次数 delay: 初始延迟(秒) backoff: 退避倍数 exceptions: 需要重试的异常类型 """ def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs): current_delay = delay last_exception = None for attempt in range(max_retries): try: return func(*args, **kwargs) except exceptions as e: last_exception = e if attempt < max_retries - 1: logger.warning( f"{func.__name__} 失败 (尝试 {attempt + 1}/{max_retries}): {e}, " f"{current_delay}秒后重试" ) time.sleep(current_delay) current_delay *= backoff else: logger.error(f"{func.__name__} 最终失败: {e}") # 所有重试都失败,抛出最后一个异常 raise last_exception return wrapper return decorator class ErrorHandler: """错误处理器""" @staticmethod def handle_ui_error(error: Exception, context: str = "") -> Dict[str, Any]: """ 处理UI操作错误 Args: error: 异常对象 context: 上下文信息 Returns: 错误信息字典 """ error_msg = str(error) # 常见错误分类 if "timeout" in error_msg.lower() or "等待" in error_msg: return { "success": False, "error": "操作超时", "type": "timeout", "context": context, "suggestion": "检查元素是否存在或增加等待时间" } elif "not found" in error_msg.lower() or "未找到" in error_msg: return { "success": False, "error": "元素未找到", "type": "element_not_found", "context": context, "suggestion": "检查选择器或使用AI Agent模式" } elif "permission" in error_msg.lower() or "权限" in error_msg: return { "success": False, "error": "权限不足", "type": "permission_denied", "context": context, "suggestion": "检查ADB权限或Root权限" } else: return { "success": False, "error": error_msg, "type": "unknown", "context": context, "suggestion": "查看日志获取详细信息" } @staticmethod def should_retry(error: Exception) -> bool: """ 判断是否应该重试 Args: error: 异常对象 Returns: 是否应该重试 """ error_msg = str(error).lower() # 这些错误可以重试 retryable_errors = [ "timeout", "等待", "network", "连接", "temporary", "临时" ] # 这些错误不应该重试 non_retryable_errors = [ "permission", "权限", "not found", "未找到", "invalid", "无效" ] for retryable in retryable_errors: if retryable in error_msg: return True for non_retryable in non_retryable_errors: if non_retryable in error_msg: return False # 默认可以重试 return True def with_error_handling(context: str = ""): """ 错误处理装饰器 Args: context: 上下文信息 """ def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs): try: result = func(*args, **kwargs) return result except Exception as e: logger.error(f"{func.__name__} 执行失败: {e}") return ErrorHandler.handle_ui_error(e, context or func.__name__) return wrapper return decorator