""" iOS Device Implementation Concrete implementation of AbstractDevice for iOS devices. """ import logging import os import time from typing import Optional, Dict, Any, List from ...core.abstract_device import AbstractDevice from ...exceptions import FATAL_EXCEPTIONS class IOSApp: """ iOS 应用的简单代理类 提供与 AndroidApp 类似的接口以兼容 EventLog 等组件。 """ def __init__(self, bundle_id: str): """ 初始化 iOS 应用 :param bundle_id: 应用的 Bundle ID """ self.bundle_id = bundle_id self.package_name = bundle_id # 兼容 Android 接口 self.main_activity = bundle_id # iOS 没有 Activity，用 bundle_id 代替 self.activities = [bundle_id] # iOS 没有 Activity 列表 def get_package_name(self) -> str: """获取包名（bundle_id）""" return self.bundle_id def get_start_intent(self): """iOS 不使用 Intent，返回 None""" return None def get_stop_intent(self): """iOS 不使用 Intent，返回 None""" return None class IOSLogcat: """ iOS 日志代理类 iOS 没有 logcat，此类提供空实现以兼容 input_policy。 """ def get_recent_lines(self) -> list: """获取最近的日志行（iOS 返回空列表）""" return [] def clear(self): """清除日志""" pass class IOSDevice(AbstractDevice): """ iOS 设备的具体实现 继承自 AbstractDevice，通过 WebDriverAgent (WDA) 实现所有 iOS 特定的设备操作。 """ def __init__(self, wda_url: str = "http://localhost:8100", bundle_id: str = None, output_dir: str = None, cv_mode: bool = False, udid: str = None, debug_mode: bool = False): """ 初始化 iOS 设备连接 :param wda_url: WDA 服务地址 (默认: http://localhost:8100) :param bundle_id: 被测应用的 Bundle ID :param output_dir: 输出目录 :param cv_mode: 是否启用 CV 模式 :param udid: 设备 UDID (可选，用于多设备场景) :param debug_mode: 是否启用调试模式，输出详细日志 """ super().__init__(output_dir=output_dir) self.wda_url = wda_url self.bundle_id = bundle_id self.cv_mode = cv_mode self.udid = udid self.debug_mode = debug_mode # 调试模式标志 # 【修复】如果启用调试模式，设置 logger 级别为 DEBUG if debug_mode: import logging self.logger.setLevel(logging.DEBUG) # 内部管理的 App 对象（兼容 input_manager） self._app = IOSApp(bundle_id) if bundle_id else None # WDA 客户端（在 connect() 中初始化，GuiAgent 等早期调用通过 get_display_info 懒加载） self._wda_client = None self._session = None # 设备信息缓存 self._device_info = None self._window_size = None # CV 模式相关缓存 self.last_screenshot_hash = None self.last_views = None self.views_cache = {} # 界面快照缓存: {dhash: views} # 状态缓存：避免 source() 的重复调用 # 每步事件循环中 get_current_state() 被调用 3 次（generate_event / EventLog.start / EventLog.stop） # 通过缓存实现一次性消费，每步只调用 1 次 source() self._cached_state = None # WDA 健康监控与卡死检测 self._wda_health_monitor = None # WDAHealthMonitor 实例（由外部注入） self._consecutive_failures = 0 # 连续 WDA 操作失败计数 self.WDA_FAILURE_THRESHOLD = 3 # 触发恢复的失败阈值 # 事件控制标志（兼容 input_manager） self.pause_sending_event = False # UTG 兼容属性 self.serial = udid or "ios_device" # 设备序列号 self.minicap = None # iOS 不使用 minicap self.adapters = {None: False} # 适配器字典，None 对应 minicap # input_policy 兼容属性 self.logcat = IOSLogcat() # 日志代理 # 流量监控：存储 pcap 抓包输出目录（由外部在启动测试后注入） # 父类将 captured_traffic_dir 定义为只读 property，用私有字段存储实际路径 self._captured_traffic_dir: Optional[str] = None # pcap 抓包进程（由外部 ios_test.py 注入，供 is_traffic_capture_running / restart_traffic_capture 使用） self._pcap_process = None # 重启 pcap 时需要的参数（由外部注入） self._pcap_restart_kwargs: dict = {} self._last_state = None # ==================== 平台信息 ==================== @property def captured_traffic_dir(self) -> Optional[str]: """pcap 抓包输出目录（覆盖父类只读 property，提供 setter）""" return self._captured_traffic_dir @captured_traffic_dir.setter def captured_traffic_dir(self, value: Optional[str]): self._captured_traffic_dir = value # ==================== pcap 进程管理 ==================== def set_pcap_process(self, process, restart_kwargs: dict = None): """注入 pcap 进程引用，供 is_traffic_capture_running / restart_traffic_capture 使用 :param process: subprocess.Popen 对象，由 ios_test.py start_pcap() 返回 :param restart_kwargs: 传给 start_pcap() 的关键字参数字典（含 bundle_id/udid/output_dir 等） """ self._pcap_process = process self._pcap_restart_kwargs = restart_kwargs or {} self.logger.info(f"pcap 进程已注入 (PID={process.pid if process else None})") def is_traffic_capture_running(self) -> bool: """检查 pcap 抓包进程是否仍在运行 iOS 通过检查注入的 _pcap_process 的存活状态来判断。 若未注入进程引用，保守地返回 True（避免 check_monitor_health 误触发重启）。 """ if self._pcap_process is None: self.logger.debug("is_traffic_capture_running: 未注入 pcap 进程，跳过检查，默认返回 True") return True is_running = self._pcap_process.poll() is None self.logger.debug(f"is_traffic_capture_running: PID={self._pcap_process.pid}, running={is_running}") return is_running def restart_traffic_capture(self, package_name: str = None): """重启 pcap 抓包进程 流程： 1. 终止旧进程 2. 用保存的 restart_kwargs 调用 ios_start.start_pcap() 重新启动 3. 更新 _pcap_process 引用 4. 重置增量读取状态（让 get_traffic_domains 从新文件头部重新读取） """ self.logger.info("[restart_traffic_capture] 开始重启 pcap 抓包...") # 1. 终止旧进程 if self._pcap_process and self._pcap_process.poll() is None: try: self._pcap_process.terminate() self._pcap_process.wait(timeout=5) self.logger.info("[restart_traffic_capture] 旧 pcap 进程已终止") except Exception as e: self.logger.warning(f"[restart_traffic_capture] 终止旧进程异常: {e}") try: self._pcap_process.kill() except Exception: pass # 若未注入重启参数，则无法重启 if not self._pcap_restart_kwargs: self.logger.warning("[restart_traffic_capture] 未注入 pcap 重启参数，跳过重启") return # 2. 重启 pcap try: from ios_start import start_pcap new_process = start_pcap(**self._pcap_restart_kwargs) if new_process: self._pcap_process = new_process self.logger.info( f"[restart_traffic_capture] 新 pcap 进程已启动 (PID={new_process.pid})" ) else: self.logger.error("[restart_traffic_capture] start_pcap 返回 None，重启失败") except Exception as e: self.logger.error(f"[restart_traffic_capture] 重启 pcap 失败: {e}") # 3. 重置增量读取状态（新抓包文件会覆盖旧文件或生成新路径） self._traffic_csv_path = None self._traffic_csv_offset = 0 self._traffic_csv_header = None self.logger.info("[restart_traffic_capture] 增量读取状态已重置") def get_platform_name(self) -> str: return "ios" # ==================== 连接管理 ==================== def set_up(self) -> None: """设置设备连接前的准备工作""" self.logger.info(f"Setting up iOS device connection to {self.wda_url}") # iOS 设备通过 WDA 连接，无需额外设置 pass def connect(self) -> bool: """连接到设备""" try: from .wda import Client self.logger.info(f"Connecting to WDA at {self.wda_url}") # 根据传入的连接参数选择连接方式： # - http: 格式 → 使用 HTTP Client 连接 # - UDID 或空字符串 → 使用 USBClient 通过 USB 连接 if self.wda_url and self.wda_url.startswith("http:"): self._wda_client = Client(self.wda_url) else: from .wda import USBClient udid = self.wda_url if self.wda_url else (self.udid or "") self._wda_client = USBClient(udid=udid) # 等待 WDA 就绪 if not self._wda_client.wait_ready(timeout=30, noprint=True): self.logger.error("WDA is not ready") return False # 获取设备信息 self._device_info = self._wda_client.status() self.logger.info(f"Connected to iOS device: {self._device_info.get('os', {}).get('version', 'unknown')}") # 获取显示信息 self.get_display_info() # 注意：不在此处启动应用 # input_policy 会先发送 KillAppEvent 杀死应用，然后再启动 # 如果在这里启动，会导致：启动 -> 杀死 -> 再启动，第一次启动是多余的 self._session = self._wda_client self.connected = True return True except Exception as e: self.logger.error(f"Failed to connect to iOS device: {e}") import traceback traceback.print_exc() return False def disconnect(self) -> None: """断开设备连接""" self.logger.info("Disconnecting from iOS device") self.connected = False # 先终止被测应用，防止设备继续执行操作 if self._wda_client and self.bundle_id: try: self.logger.info(f"Terminating app: {self.bundle_id}") self._wda_client.app_terminate(self.bundle_id) except Exception as e: self.logger.warning(f"Failed to terminate app: {e}") # 关闭 session if self._session and hasattr(self._session, 'close'): try: self._session.close() except Exception as e: self.logger.warning(f"Failed to close session: {e}") self._session = None self._wda_client = None def tear_down(self) -> None: """清理资源""" self.logger.info("Tearing down iOS device resources") # 清理临时文件 if self.output_dir: temp_dir = os.path.join(self.output_dir, "temp") if os.path.exists(temp_dir): import shutil shutil.rmtree(temp_dir) def check_connectivity(self) -> bool: """检查连接状态""" try: return self._wda_client.status() is not None except Exception as e: self.logger.error(f"Failed to check WDA connectivity: {e}") return False # ==================== 状态获取 ==================== def check_network(self, host: str = "8.8.8.8") -> bool: """ 检查 iOS 设备内部网络是否连通 :param host: 测试目标主机（iOS 不使用此参数） :return: True（假设网络可用） """ # 设备网络来自Mac共享，测试主机网络即可 try: r = subprocess.run("ping -c 1 -W 2" + host, shell=True) if "1 received" in r: return True else: return False except Exception as e: self.logger.error(f"Failed to check network connectivity: {e}") return False def get_current_state(self) -> 'IOSDeviceState': """获取当前设备状态 支持状态缓存：如果 _cached_state 不为空，直接返回缓存并清除。 每步事件循环中会被调用 3 次（generate_event / EventLog.start / EventLog.stop）， 通过缓存机制避免重复调用 source()，从每步 3 次降为 1 次。 """ # 优先返回缓存的状态（一次性消费） if self._cached_state is not None: cached = self._cached_state self._cached_state = None self.logger.debug("Using cached state (skip source() call)") return cached self.logger.debug("Getting current device state...") # 广告拦截：在获取真正的 state/views 之前，先屏蔽并退出应用下载广告 if self._wda_client: try: # 兼容普通模式和 CV 模式的统一拦截：通过“获取”文字识别广告，通过“完成”退出 if self._wda_client(label='获取').exists: self.logger.info("检测到应用下载广告，尝试点击'完成'退出...") if self._wda_client(label='完成').click_exists(timeout=2.0): self.logger.info("已成功确认广告页面并点击完成退出") import time time.sleep(1.5) # 等待动画和页面恢复 except Exception as e: self.logger.debug(f"检查或处理应用下载广告异常: {e}") current_state = None try: foreground_page = self._get_foreground_page() is_target_app = (not self.bundle_id) or (foreground_page == self.bundle_id) if not is_target_app: self.logger.warning(f"get_current_state: 当前应用 {foreground_page} 非被测应用") views = self._get_views() screenshot_path = self.take_screenshot() from .ios_device_state import IOSDeviceState current_state = IOSDeviceState( device=self, views=views, foreground_page=foreground_page, screenshot_path=screenshot_path ) except Exception as e: self.logger.warning(f"Exception in get_current_state: {e}") import traceback traceback.print_exc() if not current_state: self.logger.warning("Failed to get current state!") return current_state def get_display_info(self, refresh: bool = False) -> Dict[str, Any]: """获取显示信息 若 _wda_client 尚未初始化（如 GuiAgent 在 connect 之前调用）， 先调用 connect() 建立连接。 """ # 懒加载：如果 WDA 客户端还未建立，先连接 if self._wda_client is None: self.logger.debug("get_display_info: _wda_client 为 None，尝试懒加载连接") self.connect() if self.display_info is None or refresh: try: window_size = self._wda_client.window_size() scale = getattr(self._wda_client, 'scale', 3) self.display_info = { "width": int(window_size.width), "height": int(window_size.height), "scale": scale, "density": scale * 160 # 近似 DPI } self._window_size = window_size except Exception as e: self.logger.warning(f"Failed to get display info: {e}") self.display_info = {"width": 375, "height": 812, "scale": 2} return self.display_info # ==================== 屏幕操作 ==================== def take_screenshot(self, path: str = None) -> Optional[str]: """截取屏幕""" if self.output_dir is None and path is None: return None try: from datetime import datetime tag = datetime.now().strftime("%Y-%m-%d_%H%M%S") if path is None: local_image_dir = os.path.join(self.output_dir, "temp") if not os.path.exists(local_image_dir): os.makedirs(local_image_dir) local_image_path = os.path.join(local_image_dir, f"screen_{tag}.png") else: local_image_path = path # 使用 WDA 截图 self._wait_wda_ready() self._wda_client.screenshot(local_image_path) # TODO 将图片按照scale降采样到1/3 try: from PIL import Image with Image.open(local_image_path) as img: width, height = img.size scale = self.get_display_info().get("scale", 3) if scale > 1: # 降采样到逻辑分辨率 new_size = (int(width / scale), int(height / scale)) img = img.resize(new_size, Image.LANCZOS) img.save(local_image_path) except ImportError: self.logger.warning("PIL not installed, skipping screenshot resize") except Exception as e: self.logger.warning(f"Failed to resize screenshot: {e}") return local_image_path except FATAL_EXCEPTIONS: raise # WDAStuckError 等致命异常继续向上传播 except Exception as e: self.logger.warning(f"Failed to take screenshot: {e}") self._on_wda_failure(f"take_screenshot: {e}") return None def unlock(self) -> None: """解锁屏幕""" try: if self._wda_client.locked(): self._wda_client.unlock() time.sleep(1) except Exception as e: self.logger.warning(f"Failed to unlock: {e}") # ==================== 事件发送 ==================== def set_health_monitor(self, monitor): """设置 WDA 健康监控实例 :param monitor: WDAHealthMonitor 实例 """ self._wda_health_monitor = monitor self.logger.info("WDA 健康监控已设置") def _wait_wda_ready(self, timeout: float = 30) -> bool: """ 等待 WDA 就绪 在执行任何操作前调用此方法，确保 WDA 已就绪（之前的操作/动画已完成）。 如果 WDA 正在恢复中，自动延长超时时间。 :param timeout: 超时时间（秒） :return: WDA 是否就绪 """ if not self._wda_client: return False try: # 如果健康监控正在恢复中，延长等待超时 actual_timeout = timeout if self._wda_health_monitor and self._wda_health_monitor.is_recovering: actual_timeout = max(timeout, 60) # 恢复中至少等 60s self.logger.info(f"WDA 正在恢复中，延长等待超时 (timeout={actual_timeout}s)...") ready = self._wda_client.wait_ready(timeout=actual_timeout, noprint=True) if ready: self._on_wda_success() else: self._on_wda_failure("wait_ready 超时") return ready except Exception as e: self.logger.warning(f"Exception while waiting for WDA ready: {e}") self._on_wda_failure(str(e)) return False def _on_wda_failure(self, reason: str = ""): """WDA 操作失败时调用：累加计数，达到阈值触发异步恢复""" self._consecutive_failures += 1 self.logger.warning( f"WDA 操作失败 ({reason}), 连续失败: " f"{self._consecutive_failures}/{self.WDA_FAILURE_THRESHOLD}" ) if self._wda_health_monitor: # 检查是否已超过最大恢复次数 if self._wda_health_monitor.has_exceeded_max_attempts(): from .wda.exceptions import WDAStuckError raise WDAStuckError( f"WDA 卡死：连续 {self._wda_health_monitor.MAX_RECOVER_ATTEMPTS} " f"次恢复均失败，终止当前测试" ) # 达到失败阈值，触发异步恢复 if self._consecutive_failures >= self.WDA_FAILURE_THRESHOLD: self.logger.warning("达到失败阈值，触发 WDA 异步恢复...") self._wda_health_monitor.trigger_recovery() self._consecutive_failures = 0 # 重置计数，等待恢复结果 def _on_wda_success(self): """WDA 操作成功时调用：重置失败计数和恢复计数""" if self._consecutive_failures > 0: self.logger.info( f"WDA 恢复正常 (之前连续失败 {self._consecutive_failures} 次)" ) self._consecutive_failures = 0 if self._wda_health_monitor: self._wda_health_monitor.reset_recover_count() def send_event(self, event) -> bool: """发送输入事件""" event_str = event.get_event_str() if hasattr(event, 'get_event_str') else str(event) self.logger.debug(f"send_event: 准备发送事件 {event_str}") try: # 等待 WDA 就绪后再发送事件 self.logger.debug(f"send_event: 等待 WDA 就绪...") if not self._wait_wda_ready(timeout=10): self.logger.warning("WDA not ready, skipping event") return False self.logger.debug(f"send_event: WDA 就绪，发送事件...") event.send(self) self.logger.debug(f"send_event: 事件发送完成") return True except Exception as e: self.logger.warning(f"Failed to send event: {e}") return False def view_touch(self, x: int, y: int) -> None: """触摸指定坐标""" self.logger.debug(f"view_touch: 等待 WDA 就绪...") self._wait_wda_ready() self.logger.debug(f"view_touch: 执行点击 ({x}, {y})") self._wda_client.tap(x, y) self.logger.debug(f"view_touch: 点击完成") def view_long_touch(self, x: int, y: int, duration: float = 2.0) -> None: """长按指定坐标""" self.logger.debug(f"view_long_touch: 等待 WDA 就绪...") self._wait_wda_ready() self.logger.debug(f"view_long_touch: 执行长按 ({x}, {y}), 时长={duration}s") # WDA 使用秒为单位 self._wda_client.tap_hold(x, y, duration) self.logger.debug(f"view_long_touch: 长按完成") def view_drag(self, start_xy: tuple, end_xy: tuple, duration: float = 0.5) -> None: """拖拽操作""" self.logger.debug(f"view_drag: 等待 WDA 就绪...") self._wait_wda_ready() # 针对安卓接口单位为ms问题放缩duration if duration > 100: duration = duration / 1000 self.logger.debug(f"view_drag: 执行拖拽 {start_xy} -> {end_xy}, 时长={duration}s") self._wda_client.swipe(start_xy[0], start_xy[1], end_xy[0], end_xy[1], duration) self.logger.debug(f"view_drag: 拖拽完成") def view_set_text(self, text: str) -> None: """设置文本""" self.logger.debug(f"view_set_text: 等待 WDA 就绪...") self._wait_wda_ready() text_preview = text[:20] + '...' if len(text) > 20 else text self.logger.debug(f"view_set_text: 输入文本 '{text_preview}'") # 使用 WDA 输入文本 try: self._wda_client.send_keys(text) self.logger.debug(f"view_set_text: 文本输入完成") except Exception as e: self.logger.warning(f"Failed to set text: {e}") def _swipe_back(self): """实现iOS返回：从屏幕左边缘滑向中间""" display_info = self.get_display_info() width = display_info.get("width", 375) height = display_info.get("height", 812) start_x = 0 start_y = int(height / 2) end_x = width - int(width / 6) end_y = int(height / 2) self.logger.debug(f"_swipe_back: ({start_x}, {start_y}) -> ({end_x}, {end_y})") self._wda_client.swipe(start_x, start_y, end_x, end_y, 0.2) def key_press(self, key_code: str) -> None: """按键操作""" self.logger.debug(f"key_press: 等待 WDA 就绪...") self._wait_wda_ready() self.logger.debug(f"key_press: 执行按键 '{key_code}'") key_map = { "HOME": lambda: self._wda_client.home(), "BACK": self._swipe_back, # 使用滑动侧边实现返回 "VOLUME_UP": lambda: self._wda_client.press_volume_up(), "VOLUME_DOWN": lambda: self._wda_client.press_volume_down(), } if key_code.upper() in key_map: key_map[key_code.upper()]() self.logger.debug(f"key_press: 按键完成") else: self.logger.warning(f"Unsupported key code: {key_code}") # ==================== 应用管理 ==================== @property def app_identifier(self) -> str: """获取被测应用的唯一标识符（bundle_id）""" return self.bundle_id or "" def is_foreground(self) -> bool: """检查被测应用是否在前台""" if not self.bundle_id: return False try: # 等待 WDA 就绪（恢复期间会自动延长超时） if not self._wait_wda_ready(timeout=10): self.logger.warning("is_foreground: WDA 未就绪") return False current_app = self._wda_client.app_current() is_fg = current_app.get("bundleId") == self.bundle_id # 当 springboard 在前台时，使用 WDA 的 app_state 检查被测应用的真实状态 if current_app.get("bundleId") == "com.apple.springboard": try: # 使用 WDA 原生方法检查应用状态 # XCUIApplicationState 枚举值: # 0 = unknown (未知) # 1 = notRunning (未运行) # 2 = runningBackgroundSuspended (后台挂起) # 3 = runningBackground (后台运行) # 4 = runningForeground (前台运行) state_result = self._wda_client.app_state(self.bundle_id) app_state = state_result.get("value", 1) if app_state == 4: # 应用在前台运行（通常是被 springboard 遮挡） is_fg = True self.logger.debug(f"is_foreground: Springboard 显示，但被测应用状态=前台运行(4)") elif app_state in [2, 3]: # 应用在后台运行或挂起 is_fg = False state_name = "后台挂起" if app_state == 2 else "后台运行" self.logger.debug(f"is_foreground: 应用在{state_name}，状态值={app_state}") else: # 应用未运行或状态未知 (0 或 1) is_fg = False state_name = "未知" if app_state == 0 else "未运行" self.logger.debug(f"is_foreground: 应用{state_name}，状态值={app_state}") except Exception as e: # 如果 app_state 调用失败，保守地认为应用不在前台 self.logger.warning(f"Failed to check app state: {e}, assuming not foreground") is_fg = False self.logger.debug(f"is_foreground: 当前应用={current_app.get('bundleId')}, 目标={self.bundle_id}, 结果={is_fg}") return is_fg except FATAL_EXCEPTIONS: raise # WDAStuckError 等致命异常继续向上传播 except Exception as e: self.logger.warning(f"Failed to check foreground: {e}") self._on_wda_failure(f"is_foreground: {e}") return False def pull_back_to_app(self) -> bool: """将被测应用拉回前台""" if not self.bundle_id: return True if self.is_foreground(): return True self.logger.debug(f"pull_back_to_app: 拉回应用 {self.bundle_id}") self.logger.info(f"Pulling back to app: {self.bundle_id}") try: self._wda_client.app_activate(self.bundle_id) time.sleep(1) result = self.is_foreground() self.logger.debug(f"pull_back_to_app: 结果={result}") return result except Exception as e: self.logger.error(f"Failed to pull back to app: {e}") return False def _is_springboard_with_app_foreground(self) -> bool: """ 检查是否是 springboard 遮挡场景 当 springboard 在前台，但被测应用状态为前台运行(state=4)时， 表示出现了系统弹窗（如权限请求对话框）遮挡应用界面。 Returns: bool: True 表示 springboard 遮挡场景，False 表示正常场景 """ if not self.bundle_id: return False try: current_app = self._wda_client.app_current() if current_app.get("bundleId") == "com.apple.springboard": # Springboard 在前台，检查被测应用状态 state_result = self._wda_client.app_state(self.bundle_id) app_state = state_result.get("value", 1) if app_state == 4: # 被测应用在前台运行但被 springboard 遮挡（系统弹窗） self.logger.debug(f"检测到 springboard 遮挡场景（系统弹窗）") return True except Exception as e: self.logger.debug(f"springboard 检测失败: {e}") return False def start_app(self) -> bool: """启动被测应用""" if not self.bundle_id: self.logger.warning("No bundle_id specified, cannot start app") return False self.logger.debug(f"start_app: 启动应用 {self.bundle_id}") try: # 等待 WDA 就绪（恢复期间会自动延长超时） if not self._wait_wda_ready(timeout=10): self.logger.warning("start_app: WDA 未就绪，跳过启动") return False self._wda_client.app_launch(self.bundle_id) time.sleep(4) self.logger.debug(f"start_app: 应用启动完成") # 检测是否为游戏应用，如果是则自动启用 CV 模式 self._check_and_enable_cv_mode_if_game() return True except FATAL_EXCEPTIONS: raise # WDAStuckError 等致命异常继续向上传播 except Exception as e: self.logger.error(f"Failed to start app: {e}") self._on_wda_failure(f"start_app: {e}") return False def _check_and_enable_cv_mode_if_game(self) -> None: """ 检测应用是否为游戏，如果是则自动启用 CV 模式。 iOS 游戏（Unity/Unreal 等）的 WDA source 树通常只含容器型节点 （Application / Window / Other），没有任何标准控件（Button / Cell / TextField / StaticText 等），叶子节点 traits 会包含 'AllowsDirectInteraction'（OpenGL/Metal 渲染画布的特征）。 检测条件（满足任一即判定为游戏并启用 CV 模式）： 1. source 树中完全没有标准控件类型节点 2. 存在 traits 含 'AllowsDirectInteraction' 的节点（游戏渲染画布） """ # 已经是 CV 模式则无需检测 if self.cv_mode: return try: time.sleep(5) # 先检测并处理系统弹窗（如"允许跟踪"等权限对话框） # 系统弹窗包含 Button 等标准控件，会导致游戏误判为普通应用 MAX_ALERT_ROUNDS = 3 # 最多处理3轮弹窗（有些应用会连续弹出多个权限请求） for round_idx in range(MAX_ALERT_ROUNDS): if self._is_springboard_with_app_foreground(): try: alert_text = None try: alert_text = self._wda_client.alert.text except Exception: pass self.logger.info( f"_check_game: 检测到系统弹窗（第{round_idx + 1}轮），" f"内容: {alert_text}，尝试接受弹窗" ) self._wda_client.alert.accept() time.sleep(2) # 等待弹窗关闭和界面切换 except Exception as e: self.logger.debug(f"_check_game: 处理系统弹窗失败: {e}") break else: if round_idx > 0: self.logger.info( f"_check_game: 系统弹窗已全部处理（共{round_idx}轮）" ) break # 获取当前 UI 树 source = self._wda_client.source(format='json') if not source: self.logger.debug("_check_game: source 为空，跳过检测") return # 标准控件类型（游戏应用通常没有这些节点） STANDARD_CONTROL_TYPES = { 'Button', 'Cell', 'TextField', 'SecureTextField', 'StaticText', 'Switch', 'Link', 'SearchField', 'Slider', 'SegmentedControl', 'Picker', } found_standard_control = False found_direct_interaction = False def _traverse(node: dict) -> None: """递归遍历 source 树，统计节点特征""" nonlocal found_standard_control, found_direct_interaction if not isinstance(node, dict): return node_type = node.get('type', '') node_traits = node.get('traits', '') or '' # 检查是否包含标准控件类型 if node_type in STANDARD_CONTROL_TYPES: found_standard_control = True # 检查是否存在 AllowsDirectInteraction（游戏渲染画布特征） if 'AllowsDirectInteraction' in node_traits: found_direct_interaction = True # 递归处理子节点 for child in node.get('children', []) or []: _traverse(child) _traverse(source) # 判定逻辑：有 AllowsDirectInteraction 或完全没有标准控件，判定为游戏 if found_direct_interaction: self.logger.info( f"检测到游戏应用（source 含 AllowsDirectInteraction 渲染画布），" f"自动启用 CV 模式 [bundle_id={self.bundle_id}]" ) self.cv_mode = True elif not found_standard_control: self.logger.info( f"检测到游戏应用（source 中无任何标准控件节点），" f"自动启用 CV 模式 [bundle_id={self.bundle_id}]" ) self.cv_mode = True else: self.logger.info( f"应用包含标准控件节点，判定为普通应用，使用控件模式 " f"[bundle_id={self.bundle_id}]" ) except FATAL_EXCEPTIONS: raise except Exception as e: self.logger.debug(f"_check_game: 游戏检测失败（忽略）: {e}") def install_app(self) -> bool: """安装应用（iOS 通过其他方式安装，这里跳过）""" self.logger.info("Skipping install_app: iOS apps should be pre-installed") return True def uninstall_app(self) -> bool: """卸载应用（保留应用环境）""" self.logger.info("Skipping uninstall_app: Preserving app environment") return True # ==================== iOS 特定方法 ==================== def _get_views(self) -> Optional[List[Dict[str, Any]]]: """获取当前界面视图""" self.logger.debug(f"_get_views: 等待 WDA 就绪...") self._wait_wda_ready() if self.cv_mode: self.logger.debug(f"_get_views: 使用 CV 模式获取视图") return self._get_views_cv_mode() try: # 检查是否是 springboard 活跃但被测应用在前台的情况（系统弹窗遮挡） # 在这种情况下，source() API 可能无法获取应用的 UI 层级 # 需要使用 find_elements() 的方式获取控件 if self._is_springboard_with_app_foreground(): # 被测应用在前台运行但被 springboard 遮挡（系统弹窗） # 使用 find_elements() 方式获取控件 self.logger.debug(f"_get_views: 检测到 springboard 遮挡场景，使用 find_elements() 获取控件") return self._get_views_from_elements() # 默认方式：使用 WDA 获取界面层次结构 self.logger.debug(f"_get_views: 使用 WDA source() 获取界面层次结构") source = self._wda_client.source(format='json') views = self._parse_wda_source(source) self.logger.debug(f"_get_views: 获取到 {len(views) if views else 0} 个视图") return views except FATAL_EXCEPTIONS: raise # WDAStuckError 等致命异常继续向上传播 except Exception as e: self.logger.warning(f"Failed to get views: {e}") return None def _parse_wda_source(self, source: Dict) -> List[Dict[str, Any]]: """ 解析 WDA 返回的界面层次结构，转换为统一的 ViewDict 格式 iOS WDA 返回的结构字段： - name: 元素标识符/名称 - type: 元素类型 (Button, Cell, CollectionView, etc.) - label: 显示的文本 - value: 值 - children: 子元素列表 - rawIdentifier: 原始标识符 - rect/frame: 边界信息（可能不存在） """ views = [] temp_id_counter = [0] # 使用列表以便在递归中修改 # 【优化】在解析前获取一次前台页面，避免为每个视图都调用 WDA foreground_page = self._get_foreground_page() or "" # 获取屏幕尺寸用于默认边界 display_info = self.get_display_info() screen_width = display_info.get("width", 375) screen_height = display_info.get("height", 812) def parse_element(element: Dict, parent_id: int = -1) -> int: """递归解析元素""" if element is None: return -1 current_id = temp_id_counter[0] temp_id_counter[0] += 1 # 获取边界（可能不存在） frame = element.get("rect", element.get("frame", {})) if frame: x = frame.get("x", 0) or 0 y = frame.get("y", 0) or 0 width = frame.get("width", 0) or 0 height = frame.get("height", 0) or 0 else: # 没有边界信息，使用默认值 x, y, width, height = 0, 0, 0, 0 bounds = [[int(x), int(y)], [int(x + width), int(y + height)]] # 获取文本内容（优先使用 label，其次 value，最后 name） text = element.get("label", "") or "" value = element.get("value", "") or "" name = element.get("name", "") or "" # text = label or value or "" # 获取元素类型 element_type = element.get("type", "Unknown") or "Unknown" # 判断元素属性 is_button = element_type in ["Button","XCUIElementTypeButton"] is_scrollable = element_type in [ "Table", "TableView", "XCUIElementTypeTable", "XCUIElementTypeTableView"] # "ScrollView", "CollectionView", is_editable = element_type in [ # 标准类型名 "TextField", "SecureTextField", "TextEditor", "SearchField", # XCUIElement 类型名 "XCUIElementTypeTextField", "XCUIElementTypeSecureTextField", "XCUIElementTypeSearchField", "XCUIElementTypeTextEditor" ] # 构建 ViewDict view = { "temp_id": current_id, "parent": parent_id, "bounds": bounds, "text": text, # "content_description": label, "resource_id": element.get("rawIdentifier", "") or name, "class_name": element_type, "visible": bool(element.get("isVisible", True)) if element.get("isVisible") is not None else False, "enabled": bool(element.get("isEnabled", True)) if element.get("isEnabled") is not None else False, "clickable": is_button, # or element.get("accessible", False), "scrollable": is_scrollable, "editable": is_editable, "children": [], "source": "accessibility", # iOS 特有字段 "ios_name": name, "ios_value": value, } # 处理子元素 children = element.get("children", []) child_ids = [] if children: for child in children: if child is not None: child_id = parse_element(child, current_id) if child_id >= 0: child_ids.append(child_id) view["children"] = child_ids views.append(view) return current_id # 从根元素开始解析 if source: parse_element(source, -1) # 按 temp_id 排序 views.sort(key=lambda v: v["temp_id"]) # 生成 view_str（所有视图解析完成后），传入已获取的 foreground_page self._generate_view_strs(views, foreground_page) return views def _generate_view_strs(self, views: List[Dict[str, Any]], foreground_page: str) -> None: """生成所有视图的唯一标识字符串（仿照 Android 实现）""" for view_dict in views: self._get_view_str(view_dict, views, foreground_page) def _get_view_str(self, view_dict: Dict[str, Any], views: List[Dict[str, Any]], foreground_page: str) -> str: """ 获取视图字符串（仿照 Android 的 _get_view_str 实现） 生成规则： - Page: 当前前台页面（bundle_id） - Self: 视图自身签名 - Parents: 所有祖先视图签名（从根到父，用 // 连接） - Children: 所有子视图签名（排序后用 || 连接） - 最终用 MD5 哈希 """ if 'view_str' in view_dict and view_dict['view_str']: return view_dict['view_str'] view_signature = self._get_view_signature(view_dict) # 获取所有祖先签名 parent_strs = [] for parent_id in self._get_all_ancestors(view_dict, views): if 0 <= parent_id < len(views): parent_strs.append(self._get_view_signature(views[parent_id])) parent_strs.reverse() # 从根到父的顺序 # 获取所有子视图签名 child_strs = [] for child_id in self._get_all_children(view_dict, views): if 0 <= child_id < len(views): child_strs.append(self._get_view_signature(views[child_id])) child_strs.sort() # 【优化】使用传入的 foreground_page，而不是每次都调用 WDA # 构建视图字符串（格式与 Android 一致） view_str = "Page:%s\nSelf:%s\nParents:%s\nChildren:%s" % ( foreground_page, view_signature, "//".join(parent_strs), "||".join(child_strs) ) import hashlib view_str = hashlib.md5(view_str.encode('utf-8')).hexdigest() view_dict['view_str'] = view_str return view_str @staticmethod def _get_view_signature(view_dict: Dict[str, Any]) -> str: """ 获取视图签名（仿照 Android 的 _get_view_signature 实现） 签名格式: [class]类名[resource_id]资源ID[text]文本[enabled,checked,selected] """ if 'signature' in view_dict: return view_dict['signature'] view_text = view_dict.get('text', "None") or "None" if len(view_text) > 50: view_text = "None" # 文本过长时忽略 # 构建签名 signature = "[class]%s[resource_id]%s[text]%s[%s,%s,%s]" % ( view_dict.get('class_name', "None") or "None", view_dict.get('resource_id', "None") or "None", view_text, "enabled" if view_dict.get('enabled') else "", "checked" if view_dict.get('checked') else "", "selected" if view_dict.get('selected') else "" ) view_dict['signature'] = signature return signature @staticmethod def _get_all_ancestors(view_dict: Dict[str, Any], views: List[Dict[str, Any]]) -> List[int]: """获取所有祖先节点 ID""" result = [] parent_id = view_dict.get('parent', -1) while 0 <= parent_id < len(views): result.append(parent_id) parent_id = views[parent_id].get('parent', -1) return result @staticmethod def _get_all_children(view_dict: Dict[str, Any], views: List[Dict[str, Any]]) -> List[int]: """获取所有子节点 ID（递归）""" children = view_dict.get('children', []) if not children: return [] result = list(children) for child_id in children: if 0 <= child_id < len(views): result.extend(IOSDevice._get_all_children(views[child_id], views)) return result def _get_views_cv_mode(self) -> Optional[List[Dict[str, Any]]]: """CV 模式获取视图 - 使用统一的 ViewDict 格式""" try: from DroidBot.cv import cv # 使用已有的 take_screenshot() 获取截图路径 local_image_path = self.take_screenshot() if not local_image_path: self.logger.error("_get_views_cv_mode: 截图失败") return None img = cv.load_image_from_path(local_image_path) current_screenshot_hash = cv.calculate_dhash(img) # 模糊匹配缓存（相似度 > 88% 视为命中） best_match_views = None max_distance = len(current_screenshot_hash) * 4 for cached_hash in self.views_cache: hamming_distance = cv.dhash_hamming_distance(current_screenshot_hash, cached_hash) similarity = 1.0 - (hamming_distance / max_distance) if similarity > 0.88: best_match_views = self.views_cache[cached_hash] break if best_match_views: self.logger.info(f"View cache hit! Similarity: {similarity:.2f}") self.last_screenshot_hash = current_screenshot_hash self.last_views = best_match_views try: os.remove(local_image_path) except: pass return best_match_views # 未命中缓存，调用 CV 识别控件 cv_views = cv.find_views(img) # 构建根视图 display_info = self.get_display_info() width = display_info.get('width', 390) height = display_info.get('height', 844) root_view = { "class_name": "CVViewRoot", "bounds": [[0, 0], [width, height]], "enabled": True, "visible": True, "clickable": False, "scrollable": False, "editable": False, "temp_id": 0, "children": [], "text": "", "source": "cv", "resource_id": "", "view_str": "cv_root", "parent": -1, } # 重新分配 temp_id 并设置父子关系 views = [root_view] for idx, view in enumerate(cv_views): view["temp_id"] = idx + 1 view["parent"] = 0 views.append(view) root_view["children"] = list(range(1, len(views))) self.last_screenshot_hash = current_screenshot_hash self.last_views = views # 更新缓存（LRU，上限 50 个） self.views_cache[current_screenshot_hash] = views if len(self.views_cache) > 50: self.views_cache.pop(next(iter(self.views_cache))) try: os.remove(local_image_path) except: pass return views except FATAL_EXCEPTIONS: raise except Exception as e: self.logger.error(f"Failed to get views using CV mode: {e}") return None def _get_views_from_elements(self, fetch_all: bool = False) -> Optional[List[Dict[str, Any]]]: """ 使用 find_elements() 获取界面控件（用于 springboard 场景） 当 springboard 活跃但被测应用在前台(state=4)时使用此方法。 通过并行请求获取所有可交互元素的属性。 Args: fetch_all: 是否获取所有元素的完整属性。默认 False，此时当元素数量 > 5 时， 只获取第一个和最后一个元素的完整属性，其他元素只获取 label 和 value属性 Returns: 包含所有元素的视图列表 """ self.logger.debug(f"_get_views_from_elements: 使用 find_elements() 获取界面控件 (fetch_all={fetch_all})") try: import concurrent.futures # 查找所有常见的可交互元素类型 element_types = ['Button','TextField','SearchField'] # 记录元素及其类型 all_elements = [] # [(element, element_type), ...] for elem_type in element_types: try: elements = self._wda_client(type=elem_type).find_elements() # 将元素和类型一起存储 for elem in elements: all_elements.append((elem, elem_type)) self.logger.debug(f"找到 {len(elements)} 个 {elem_type} 元素") except Exception as e: self.logger.debug(f"查找 {elem_type} 失败: {e}") continue if not all_elements: self.logger.warning("未找到任何界面元素") return [] self.logger.debug(f"总共找到 {len(all_elements)} 个元素，开始并行获取属性...") # 并行获取元素属性（根据 fetch_all 参数和元素类型决定是否获取全部属性） element_properties = self._fetch_element_properties_parallel(all_elements, fetch_all=fetch_all) self.logger.debug(f"成功获取 {len(element_properties)} 个元素的属性") # 转换为标准视图格式 views = self._parse_elements_to_views(element_properties) return views except Exception as e: self.logger.error(f"Failed to get views from elements: {e}") import traceback traceback.print_exc() return None def _fetch_element_properties_parallel(self, elements: List, fetch_all: bool = False) -> List[Dict[str, Any]]: """ 并行获取多个 WDA Element 的属性 使用 ThreadPoolExecutor 并行发送 HTTP 请求获取每个元素的属性。 WDA 基于 HTTP，支持并发请求。 优化策略： - Button、TextField、SecureTextField、SearchField 始终获取完整属性 - 其他元素类型： - 当 fetch_all=False 且总元素数量 > 5 时，只获取 label 和 value 属性 - 当 fetch_all=True 或总元素数量 <= 5 时，获取所有属性 Args: elements: 元组列表 [(WDA Element, element_type), ...] fetch_all: 是否获取所有元素的完整属性 Returns: 包含所有元素属性的字典列表 """ import concurrent.futures from concurrent.futures import ThreadPoolExecutor def fetch_full_properties(elem, idx: int) -> Optional[Dict[str, Any]]: """获取单个元素的所有属性""" try: props = { 'element_index': idx, # 记录原始索引 'element_id': elem.id, 'bounds': elem.bounds, # Rect 对象 'label': elem.label, 'accessible': elem.accessible, 'name': elem.name, 'value': elem.value, 'enabled': bool(elem.enabled), 'visible': bool(elem.visible), } return props except Exception as e: self.logger.debug(f"获取元素完整属性失败 (idx={idx}): {e}") return None def fetch_minimal_properties(elem, idx: int) -> Optional[Dict[str, Any]]: """只获取元素的 label 和 value 属性（最小开销）""" try: props = { 'element_index': idx, # 记录原始索引 'label': elem.label, 'value': elem.value, # 只获取 value，减少 HTTP 请求 } return props except Exception as e: self.logger.debug(f"获取元素 label 和 value 失败 (idx={idx}): {e}") return None elements_count = len(elements) # 定义始终获取完整属性的元素类型 always_full_types = {'Button', 'TextField', 'SecureTextField', 'SearchField'} # 决定获取策略 if fetch_all or elements_count <= 5: # 获取所有元素的完整属性 self.logger.debug(f"获取策略: 全部元素完整属性 (count={elements_count})") fetch_targets = [(elem, idx, True, elem_type) for idx, (elem, elem_type) in enumerate(elements)] else: # 优化模式：根据元素类型决定是否获取完整属性 self.logger.debug(f"获取策略: 根据类型决定属性获取深度 (count={elements_count})") fetch_targets = [] for idx, (elem, elem_type) in enumerate(elements): # Button、TextField、SecureTextField、SearchField 始终获取完整属性 # 其他元素类型只获取 label 和 value is_full = elem_type in always_full_types fetch_targets.append((elem, idx, is_full, elem_type)) # 使用线程池并行获取 # 优化：根据策略调整并发数，最小开销请求可以更高并发 max_workers = min(50 if not fetch_all else 30, elements_count) properties_list = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交任务 futures = {} for elem, idx, is_full, elem_type in fetch_targets: if is_full: future = executor.submit(fetch_full_properties, elem, idx) else: future = executor.submit(fetch_minimal_properties, elem, idx) futures[future] = (idx, elem_type) # 收集结果（使用 as_completed 提高响应速度） for future in concurrent.futures.as_completed(futures): try: props = future.result(timeout=5) if props: properties_list.append(props) except Exception as e: idx, elem_type = futures[future] self.logger.debug(f"Future 执行失败 (idx={idx}, type={elem_type}): {e}") continue # 按原始索引排序，保持元素顺序 properties_list.sort(key=lambda p: p.get('element_index', 0)) return properties_list def _parse_elements_to_views(self, element_properties: List[Dict]) -> List[Dict[str, Any]]: """ 将 Element 属性列表转换为标准的 View 格式 类似于 _parse_wda_source，但用于从 find_elements() 返回的元素。 注意：find_elements() 不提供层级关系，所以 parent 和 children 都为空。 支持两种模式： 1. 完整属性模式：包含 bounds, label, name, value 等所有属性 2. 最小属性模式：只包含 value 属性（用于优化性能） Args: element_properties: 包含元素属性的字典列表 Returns: 标准格式的视图列表 """ views = [] foreground_page = self._get_foreground_page() or "" for idx, props in enumerate(element_properties): try: # 检查是否是最小属性模式（只有 value） is_minimal = 'bounds' not in props and 'label' not in props if is_minimal: # 最小属性模式：只有 value，使用默认值填充其他字段 # 这些元素主要用于生成页面唯一标识，不用于交互 value = props.get('value', '') or '' view = { "temp_id": props.get('element_index', idx), "parent": -1, "bounds": [[0, 0], [0, 0]], # 默认边界 "text": value, # 使用 value 作为 text，确保不同元素有不同的签名 "resource_id": "", "class_name": "Unknown", "visible": False, # 标记为不可见（表示未完整加载） "enabled": False, "clickable": False, "scrollable": False, "editable": False, "children": [], "source": "find_elements_minimal", # iOS 特有字段 "ios_name": "", "ios_value": value, # 保留 value 用于页面标识 "accessible": False, } else: # 完整属性模式 # 获取边界信息 bounds_rect = props.get('bounds') if bounds_rect: x, y = bounds_rect.x, bounds_rect.y w, h = bounds_rect.width, bounds_rect.height bounds = [[int(x), int(y)], [int(x + w), int(y + h)]] else: bounds = [[0, 0], [0, 0]] # 获取文本内容 label = props.get('label', '') or '' name = props.get('name', '') or '' value = props.get('value', '') or '' text = label or name or '' # 判断元素属性 is_button = 'Button' in name is_editable = 'TextField' in name or 'SearchField' in name # 构建 ViewDict view = { "temp_id": props.get('element_index', idx), "parent": -1, # find_elements() 不提供层级信息 "bounds": bounds, "text": text, "resource_id": props.get('element_id', ''), "class_name": name, "visible": bool(props.get('visible', True)), "enabled": bool(props.get('enabled', True)), "clickable": is_button, "scrollable": False, # 无法从单个元素判断 "editable": is_editable, "children": [], # find_elements() 不提供层级信息 "source": "find_elements", # iOS 特有字段 "ios_name": name, "ios_value": value, "accessible": props.get('accessible', False), } views.append(view) except Exception as e: self.logger.debug(f"解析元素 {idx} 失败: {e}") continue # 生成 view_str（即使是最小属性模式，也能参与页面标识生成） if views: self._generate_view_strs(views, foreground_page) return views def _get_foreground_page(self) -> Optional[str]: """获取前台页面标识""" try: current_app = self._wda_client.app_current() bundle_id = current_app.get("bundleId", "") # iOS 没有 Activity 概念，使用 bundle_id 作为页面标识 return bundle_id except: return None def get_device_info(self) -> Dict[str, Any]: """获取设备信息""" if self._device_info is None: try: self._device_info = self._wda_client.status() except: self._device_info = {} return self._device_info def get_ios_version(self) -> str: """获取 iOS 版本""" info = self.get_device_info() return info.get("os", {}).get("version", "unknown") def get_model_name(self) -> str: """获取设备型号""" try: device_info = self._wda_client.device_info() return device_info.get("name", "iPhone") except: return "iPhone" def get_model_number(self) -> str: """获取设备型号（兼容 UTG 接口）""" return self.get_model_name() def get_sdk_version(self) -> int: """获取 SDK 版本（兼容 UTG 接口） iOS 没有 SDK 版本概念，返回 iOS 主版本号 """ try: version = self.get_ios_version() # 返回主版本号，如 "18.5" -> 18 return int(version.split(".")[0]) except: return 0 def get_app_pid(self, app_identifier: str) -> int: """获取应用进程 ID（兼容 input_policy 接口） iOS 通过 WDA 无法直接获取 PID，返回 -1 """ # iOS 不支持直接获取 PID return -1 def get_traffic_domains(self, remote_dir: str) -> Optional[str]: """ 获取 iOS 流量域名日志，对应 TrafficMonitor 的接口。 iOS 不通过 ADB，而是直接读取 pcap 工具在本地生成的 *.flows.csv 文件（TargetDomain 列）并转换为 TrafficMonitor._parse_content() 期望的格式： 包名,应用名,域名 性能优化：记录上次读取偏移量，每次只读新增行（增量读取）。 CSV 文件切换时自动重置偏移量。 :param remote_dir: 未使用（保留接口兼容），实际路径根据 self.output_dir 确定 :return: 多行文本 \"包名,应用名,域名\" ，无可用数据时返回 None """ import csv as csv_module # 确定搜索目录，按优先级排列 # 实际布局： # captured_traffic_dir = .../com_monopoly_.../ (由 ios_test.py 注入) # device.output_dir = .../com_monopoly_.../droidbot/ # flows.csv = .../com_monopoly_.../traffic/xxx.flows.csv search_dirs = [] if self.captured_traffic_dir and os.path.isdir(self.captured_traffic_dir): # 优先：captured_traffic_dir/traffic/ 子目录（pcap 实际写入位置） sub_traffic = os.path.join(self.captured_traffic_dir, "traffic") if os.path.isdir(sub_traffic): search_dirs.append(sub_traffic) # 备用：captured_traffic_dir 本身 search_dirs.append(self.captured_traffic_dir) if self.output_dir: # output_dir 是 droidbot 子目录，pcap traffic 在其父目录的 traffic/ 下 parent_traffic = os.path.join(os.path.dirname(self.output_dir), "traffic") if os.path.isdir(parent_traffic) and parent_traffic not in search_dirs: search_dirs.append(parent_traffic) traffic_dir = os.path.join(self.output_dir, "traffic") if os.path.isdir(traffic_dir) and traffic_dir not in search_dirs: search_dirs.append(traffic_dir) if self.output_dir not in search_dirs: search_dirs.append(self.output_dir) # 最低优先级备用 if not search_dirs: self.logger.debug("get_traffic_domains: 无有效搜索目录") return None # 查找最新的 *.flows.csv（同时递归搜索一层子目录，兼容 pcap 在子目录写文件的情形） csv_path = None latest_mtime = -1 for d in search_dirs: try: for fname in os.listdir(d): fpath = os.path.join(d, fname) if fname.endswith(".flows.csv"): mtime = os.path.getmtime(fpath) if mtime > latest_mtime: latest_mtime = mtime csv_path = fpath elif os.path.isdir(fpath): # 递归搜索一层子目录（兼容 pcap 放在 traffic/ 子目录） try: for sub_fname in os.listdir(fpath): if sub_fname.endswith(".flows.csv"): sub_fpath = os.path.join(fpath, sub_fname) mtime = os.path.getmtime(sub_fpath) if mtime > latest_mtime: latest_mtime = mtime csv_path = sub_fpath except Exception: pass except Exception: pass if not csv_path: self.logger.debug("get_traffic_domains: 未找到 .flows.csv 文件") return None # 懒初始化增量读取状态 if not hasattr(self, '_traffic_csv_path'): self._traffic_csv_path = None self._traffic_csv_offset = 0 # 字节偏移量 self._traffic_csv_header = None # 列名列表 # CSV 文件切换时重置状态 if csv_path != self._traffic_csv_path: self._traffic_csv_path = csv_path self._traffic_csv_offset = 0 self._traffic_csv_header = None self.logger.debug(f"get_traffic_domains: 切换到新 CSV: {os.path.basename(csv_path)}") # 增量读取：从上次偏移量开始，只读新增内容 lines = [] try: with open(csv_path, "r", encoding="utf-8", errors="ignore") as f: # 第一次读取：解析表头 if self._traffic_csv_header is None: header_line = f.readline() if not header_line: return None self._traffic_csv_header = [h.strip() for h in header_line.split(',')] self._traffic_csv_offset = f.tell() else: # 跳到上次读取位置 f.seek(self._traffic_csv_offset) # 只读新增行 new_content = f.read() new_offset = f.tell() if not new_content.strip(): return None # 无新增内容 # 更新偏移量 self._traffic_csv_offset = new_offset # 解析新增行 header = self._traffic_csv_header try: domain_idx = header.index('TargetDomain') except ValueError: domain_idx = -1 try: bundle_idx = header.index('BundleID') except ValueError: bundle_idx = -1 try: appname_idx = header.index('AppName') except ValueError: appname_idx = -1 for raw_line in new_content.splitlines(): raw_line = raw_line.strip() if not raw_line: continue # 简单 CSV 拆分（无引号场景） parts = [p.strip() for p in raw_line.split(',')] domain = parts[domain_idx] if domain_idx >= 0 and domain_idx < len(parts) else '' if not domain: continue pkg = (parts[bundle_idx] if bundle_idx >= 0 and bundle_idx < len(parts) else '') or (self.bundle_id or '') app_name = parts[appname_idx] if appname_idx >= 0 and appname_idx < len(parts) else '' # 确保 pkg 和 app_name 本身不含逗号，使输出行严格为三列 # _parse_content 依赖 parts[2] 取 domain，若中间字段含逗号会导致偏移 pkg_safe = pkg.replace(',', '_') app_name_safe = app_name.replace(',', '_') lines.append(f"{pkg_safe},{app_name_safe},{domain}") except Exception as e: self.logger.warning(f"get_traffic_domains: 读取 flows.csv 失败: {e}") return None if not lines: return None self.logger.debug( f"get_traffic_domains: 从 {os.path.basename(csv_path)} " f"读取 {len(lines)} 条新增流量记录" ) return "\n".join(lines)