# coding: utf-8 """ iOS 自动化操作基础类 提供统一的基础设施用于实现各种 iOS 自动化任务，包括： - WDA 客户端管理 - 统一的元素查找和操作 - 弹窗处理 - 日志记录 - 会话管理 """ from typing import List, Optional, Tuple, Dict, Any import time import sys # 使用绝对导入替代相对导入 try: from utils_ios.wda import wda except ImportError: # 如果在 DroidBot 包外运行，尝试直接导入 import wda try: import logzero if not (hasattr(sys.stdout, 'isatty') and sys.stdout.isatty()): log_format = '[%(levelname)1.1s %(asctime)s %(module)s:%(lineno)d] %(message)s' logzero.setup_default_logger(formatter=logzero.LogFormatter(fmt=log_format)) logger = logzero.logger except ImportError: import logging logger = logging.getLogger(__name__) class ElementFinder: """统一的控件查找器 提供多种方式查找和操作 iOS 界面元素 在每次操作前自动处理 Alert 弹窗 """ def __init__(self, client: wda.Client, popup_handler: Optional['PopupHandler'] = None, health_monitor=None): """ 初始化控件查找器 Args: client: WDA Client 实例 popup_handler: PopupHandler 实例（可选，用于自动处理 Alert） health_monitor: WDAHealthMonitor 实例（可选，用于在 WDA 卡死时快速失败） """ self.client = client self.popup_handler = popup_handler self.health_monitor = health_monitor def _check_health_fast_fail(self): """检查 WDA 健康状态，异常时等待恢复后继续，恢复失败才抛出异常 原行为：检测到 WDA 下线 → 立刻抛 RuntimeError 新行为：检测到 WDA 下线 → 调用 trigger_recovery_and_wait 阻塞等待恢复 → 恢复成功则透明继续（调用方重试操作） → 恢复失败（超次数/超时）才抛 RuntimeError """ if self.health_monitor and not self.health_monitor.check_health(timeout=2.0): logger.warning("WDA 检测到不健康，等待自动恢复...") recovered = self.health_monitor.trigger_recovery_and_wait(timeout=120) if not recovered: raise RuntimeError("WDA 恢复失败，放弃当前任务") def _auto_handle_alert(self) -> int: """ 自动处理 Alert（如果配置了 popup_handler） Returns: 处理的 Alert 数量 """ if self.popup_handler: return self.popup_handler.handle_alert(max_attempts=5) return 0 def find_and_click(self, labels: List[str], element_type: str = 'Button', timeout: float = 1.0, auto_handle_alert: bool = True) -> Tuple[bool, Optional[str]]: """ 查找并点击控件（操作前自动处理 Alert） Args: labels: 要查找的标签列表（按优先级） element_type: 控件类型（如 'Button', 'StaticText' 等） timeout: 每个标签的查找超时时间 auto_handle_alert: 是否在操作前自动处理 Alert Returns: (是否成功, 匹配的标签) """ # 操作前先处理 Alert if auto_handle_alert: alerts_handled = self._auto_handle_alert() if alerts_handled > 0: logger.debug(f"操作前处理了 {alerts_handled} 个 Alert") self._check_health_fast_fail() for label in labels: try: # 使用线程超时保护 click_exists，避免 WDA HTTP 请求阻塞过久 # WDA 底层 HTTP_TIMEOUT 默认 180s，click_exists 的 timeout 仅控制元素等待 import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit( lambda l=label: self.client(label=l, type=element_type).click_exists(timeout=timeout) ) clicked = future.result(timeout=timeout + 10) if clicked: logger.debug(f"点击成功: {label} ({element_type})") return True, label except concurrent.futures.TimeoutError: logger.debug(f"查找控件超时 (>{timeout + 10:.1f}s): {label}") continue except Exception as e: logger.debug(f"查找控件异常: {label} - {e}") continue logger.debug(f"未找到控件: {labels}") return False, None def check_exists(self, texts: List[str], check_types: List[str] = None, auto_handle_alert: bool = True) -> Tuple[bool, Optional[str]]: """ 检查文本是否存在（检查前自动处理 Alert） Args: texts: 要检查的文本列表 check_types: 检查类型列表，如 ['label', 'name', 'value'] auto_handle_alert: 是否在检查前自动处理 Alert Returns: (是否存在, 匹配的文本) """ # 检查前先处理 Alert if auto_handle_alert: alerts_handled = self._auto_handle_alert() if alerts_handled > 0: logger.debug(f"检查前处理了 {alerts_handled} 个 Alert") self._check_health_fast_fail() if check_types is None: check_types = ['label', 'name', 'value'] for text in texts: for check_type in check_types: try: if check_type == 'label' and self.client(labelContains=text).exists: return True, text elif check_type == 'name' and self.client(nameContains=text).exists: return True, text elif check_type == 'value' and self.client(valueContains=text).exists: return True, text except Exception: pass return False, None def wait_for_element(self, labels: List[str], element_type: str = 'Button', timeout: float = 10.0, auto_handle_alert: bool = True) -> Tuple[bool, Optional[str]]: """ 等待控件出现（等待期间自动处理 Alert） Args: labels: 要等待的标签列表 element_type: 控件类型 timeout: 总超时时间 auto_handle_alert: 是否在等待期间自动处理 Alert Returns: (是否找到, 匹配的标签) """ start = time.time() while time.time() - start < timeout: # 等待期间处理 Alert if auto_handle_alert: self._auto_handle_alert() self._check_health_fast_fail() for label in labels: if self.client(label=label, type=element_type).exists: logger.debug(f"找到元素: {label} ({element_type})") return True, label time.sleep(0.5) logger.debug(f"等待元素超时: {labels}") return False, None class PopupHandler: """统一弹窗处理器 处理系统 Alert 和常规弹窗，提供自动检查和处理机制 """ # 常见的允许/确认按钮文案 ALLOW_BUTTONS = [ # 中文 "允许", "好", "确定", "始终允许", "使用App时允许", "允许一次", "仅在使用应用期间", "始终", "15分钟后需要", # 英文 # "Allow", "OK", "Yes", "Always Allow", "Allow While Using App", # "Allow Once", "While Using the App", "Always", ] # 常见的拒绝/取消按钮文案 DENY_BUTTONS = [ # 中文 "不允许", "取消", "稍后", "以后", "暂不", # 英文 # "Don't Allow", "Cancel", "Later", "Not Now", "Deny", ] def __init__(self, client: wda.Client, finder: ElementFinder, default_alert_strategy: str = "allow"): """ 初始化弹窗处理器 Args: client: WDA Client 实例 finder: ElementFinder 实例 default_alert_strategy: 默认的 Alert 处理策略 - "allow" 或 "deny" """ self.client = client self.finder = finder self.default_alert_strategy = default_alert_strategy def handle_alert(self, strategy: str = None, max_attempts: int = 5) -> int: """ 处理系统 Alert 弹窗（支持多次尝试） 使用 alert.accept()/alert.dismiss() 直接处理， 通过返回的 status 字段判断是否成功（status=0 表示成功） Args: strategy: 处理策略 - "allow" 或 "deny"，None 则使用默认策略 max_attempts: 最大尝试次数 Returns: 处理的 Alert 数量 """ if strategy is None: strategy = self.default_alert_strategy handled_count = 0 for _ in range(max_attempts): try: alert = self.client.alert # 使用线程超时保护 alert.exists 检查，避免 WDA 阻塞过久 import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(lambda: alert.exists) try: alert_exists = future.result(timeout=30) except concurrent.futures.TimeoutError: logger.debug("检查 Alert 超时 (>30s)，跳过") break if not alert_exists: break # 直接使用 accept/dismiss 方法 try: if strategy == "allow": result = alert.accept() else: result = alert.dismiss() # 检查返回状态 # 正常返回: {'value': None, 'sessionId': '...', 'status': 0} # status=0 表示成功 if isinstance(result, dict) and result.get('status') == 0: handled_count += 1 logger.debug(f"处理 Alert: 使用 {strategy} 方法成功") time.sleep(0.3) continue else: # status 非 0，可能失败了 logger.debug(f"处理 Alert 返回状态: {result}") break except Exception as e: # 如果出现异常，说明可能没有 Alert 或处理失败 logger.debug(f"处理 Alert 异常: {e}") break except Exception as e: logger.debug(f"检查 Alert 异常: {e}") break return handled_count def dismiss_popups(self, max_attempts: int = 3, button_list: List[str] = None, auto_handle_alert = False): """ 自动处理常规弹窗 Args: max_attempts: 最多尝试处理的弹窗数量 button_list: 自定义按钮列表，默认使用 ALLOW_BUTTONS auto_handle_alert: 是否先进行alert检查和处理，默认关闭 """ if button_list is None: button_list = self.ALLOW_BUTTONS for _ in range(max_attempts): clicked, label = self.finder.find_and_click(button_list, timeout=0.5, auto_handle_alert=auto_handle_alert) if not clicked: break logger.debug(f"关闭弹窗: {label}") time.sleep(0.3) class IOSAutomationBase: """iOS 自动化操作基础类 提供统一的初始化、会话管理和基础操作方法 子类可以继承此类来实现具体的自动化任务 自动处理 Alert 机制： - 所有控件操作前自动检查并处理 Alert - 可通过 default_alert_strategy 配置默认处理策略 """ def __init__(self, server_url: str, default_alert_strategy: str = "allow", health_monitor=None, **kwargs): """ 初始化基础类 Args: server_url: WDA Server URL default_alert_strategy: 默认的 Alert 处理策略 - "allow" 或 "deny" health_monitor: WDAHealthMonitor 实例（可选） **kwargs: 子类可以传入额外的配置参数 """ self.server_url = server_url self.default_alert_strategy = default_alert_strategy self.health_monitor = health_monitor self.config = kwargs self.client: Optional[wda.Client] = None self.finder: Optional[ElementFinder] = None self.popup_handler: Optional[PopupHandler] = None def _init_session(self): """初始化 WDA 会话和工具类（ElementFinder 自动关联 PopupHandler）""" # 根据传入的连接参数选择连接方式： # - http: 格式 → 使用 HTTP Client 连接 # - UDID 或空字符串 → 使用 USBClient 通过 USB 连接 if self.server_url and self.server_url.startswith("http:"): self.client = wda.Client(self.server_url) else: udid = self.server_url or "" self.client = wda.USBClient(udid=udid) # 先创建 PopupHandler self.popup_handler = PopupHandler(self.client, None, self.default_alert_strategy) # 创建 ElementFinder 并关联 PopupHandler（用于自动处理 Alert） self.finder = ElementFinder(self.client, self.popup_handler, self.health_monitor) # 更新 PopupHandler 的 finder 引用 self.popup_handler.finder = self.finder logger.debug(f"会话初始化完成: {self.server_url}, Alert策略: {self.default_alert_strategy}") def _check_wda_ready(self, timeout: float = 30.0) -> bool: """ 检查 WDA 是否就绪 Args: timeout: 等待超时时间 Returns: WDA 是否就绪 """ try: if not self.client.wait_ready(timeout=timeout): logger.error("WDA 未就绪") return False logger.debug("WDA 已就绪") return True except Exception as e: logger.error(f"检查 WDA 状态失败: {e}") return False def _unlock_device(self) -> bool: """ 解锁设备（如果已锁定） Returns: 是否成功解锁 """ try: if self.client.locked(): self.client.unlock() logger.info("设备已解锁") return True return True except Exception as e: logger.warning(f"解锁设备失败: {e}") return False def _launch_app(self, bundle_id: str) -> bool: """ 启动应用 Args: bundle_id: 应用的 Bundle ID Returns: 是否成功启动 """ try: self.client.session(bundle_id) logger.info(f"启动应用: {bundle_id}") time.sleep(2) # 等待应用启动 return True except Exception as e: logger.error(f"启动应用失败: {bundle_id} - {e}") return False def _open_url(self, url: str, wait_time: float = 5.0) -> bool: """ 打开 URL（如 URL Scheme） Args: url: 要打开的 URL wait_time: 打开后等待时间 Returns: 是否成功打开 """ try: self.client.open_url(url) logger.debug(f"打开 URL: {url}") self.client.wait_ready(wait_time) return True except Exception as e: logger.error(f"打开 URL 失败: {url} - {e}") return False def execute(self, *args, **kwargs) -> bool: """ 执行自动化任务的主方法 子类必须实现此方法来定义具体的任务流程 Returns: 任务是否成功完成 """ raise NotImplementedError("子类必须实现 execute() 方法")