import subprocess import threading import time import os import sys from typing import Tuple, Optional from Entity.Variables import WdaAppBundleId class IOSActivator: """ 专门给 iOS17+ 设备用的 go-ios 激活器： - 外部先探测 WDA，不存在时再调用 activate_ios17 - 内部流程：tunnel start -> pair(等待成功) -> image auto -> runwda """ def __init__( self, ios_path: Optional[str] = None, pair_timeout: int = 60, # 配对最多等多久 pair_retry_interval: int = 3, # 每次重试间隔 ): """ :param ios_path: ios.exe 的绝对路径，例如 E:\\code\\Python\\iOSAi\\resources\\ios.exe 如果为 None，则自动从项目的 resources 目录中寻找 ios.exe """ # ==== 统一获取 resources 目录（支持源码运行 + Nuitka EXE） ==== if "__compiled__" in globals(): # 被 Nuitka 编译后的 exe 运行时 base_dir = os.path.dirname(sys.executable) # exe 所在目录 else: # 开发环境，直接跑 .py cur_file = os.path.abspath(__file__) # 当前 .py 文件所在目录 base_dir = os.path.dirname(os.path.dirname(cur_file)) # 回到项目根 iOSAi resource_dir = os.path.join(base_dir, "resources") # 如果外部没有显式传 ios_path，就用 resources/ios.exe if ios_path is None or ios_path == "": ios_path = os.path.join(resource_dir, "ios.exe") self.ios_path = ios_path self.pair_timeout = pair_timeout self.pair_retry_interval = pair_retry_interval self._lock = threading.Lock() # go-ios tunnel 的后台进程 self._tunnel_proc: Optional[subprocess.Popen] = None # Windows: 避免弹黑框 self._creationflags = 0 self._startupinfo = None # ⭐ 新增：统一控制窗口隐藏 if os.name == "nt": try: self._creationflags = subprocess.CREATE_NO_WINDOW # type: ignore[attr-defined] except Exception: self._creationflags = 0 # ⭐ 用 STARTUPINFO + STARTF_USESHOWWINDOW 彻底隐藏窗口 si = subprocess.STARTUPINFO() si.dwFlags |= subprocess.STARTF_USESHOWWINDOW si.wShowWindow = 0 # SW_HIDE self._startupinfo = si # ======================= # 基础执行封装 # ======================= def _run( self, args, desc: str = "", timeout: Optional[int] = None, check: bool = True, ) -> Tuple[int, str, str]: """ 同步执行 ios.exe，等它返回。 :return: (returncode, stdout, stderr) """ cmd = [self.ios_path] + list(args) try: proc = subprocess.run( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=timeout, creationflags=self._creationflags, startupinfo=self._startupinfo, # ⭐ 关键：隐藏窗口 ) except subprocess.TimeoutExpired: if check: raise return -1, "", "timeout" out = proc.stdout or "" err = proc.stderr or "" if check and proc.returncode != 0: raise RuntimeError(f"[ios] 命令失败({desc}), returncode={proc.returncode}") return proc.returncode, out, err def _spawn_tunnel(self) -> None: with self._lock: if self._tunnel_proc is not None and self._tunnel_proc.poll() is None: print("[ios] tunnel 已经在运行，跳过重新启动") return cmd = [self.ios_path, "tunnel", "start"] print("[ios] 启动 go-ios tunnel: %s", " ".join(cmd)) try: proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, creationflags=self._creationflags, startupinfo=self._startupinfo, # ⭐ 关键：隐藏窗口 ) except Exception as e: # 这里改成 warning，并且直接返回，不往外抛 print("[ios] 启动 tunnel 失败(忽略): %s", e) return self._tunnel_proc = proc print("[ios] tunnel 启动成功, PID=%s", proc.pid) threading.Thread( target=self._drain_process_output, args=(proc, "tunnel"), daemon=True, ).start() def _drain_process_output(self, proc: subprocess.Popen, name: str): """简单把后台进程的输出打到日志里，避免缓冲区阻塞。""" try: if proc.stdout: for line in proc.stdout: line = line.rstrip() print(line) except Exception as e: print("[ios][%s] 读取 stdout 异常: %s", name, e) try: if proc.stderr: for line in proc.stderr: line = line.rstrip() if line: print("[ios][%s][stderr] %s", name, line) except Exception as e: print("[ios][%s] 读取 stderr 异常: %s", name, e) # ======================= # 具体步骤封装 # ======================= def _pair_until_success(self, udid: str) -> None: """ 调用 `ios --udid pair`，直到成功或者超时。 成功条件：stdout 中出现 "Successfully paired" """ deadline = time.time() + self.pair_timeout attempt = 0 while True: attempt += 1 print("[ios] 开始配对设备(%s)，第 %d 次尝试", udid, attempt) rc, out, err = self._run( ["--udid", udid, "pair"], desc=f"pair({udid})", timeout=20, check=False, ) text = (out or "") + "\n" + (err or "") if "Successfully paired" in text: print("[ios] 设备 %s 配对成功", udid) return if time.time() >= deadline: raise RuntimeError(f"[ios] 设备 {udid} 在超时时间内配对失败") time.sleep(self.pair_retry_interval) def _mount_dev_image(self, udid: str) -> None: """ `ios --udid image auto` 挂载开发者镜像。 成功条件：输出里出现 "success mounting image" 或 "there is already a developer image mounted" 之类的提示。 挂载失败现在只打 warning，不再抛异常，避免阻断后续 runwda。 """ print("[ios] 开始为设备 %s 挂载开发者镜像 (image auto)", udid) rc, out, err = self._run( ["--udid", udid, "image", "auto"], desc=f"image auto({udid})", timeout=300, check=False, ) text = (out or "") + "\n" + (err or "") text_lower = text.lower() # 这些都当成功处理 success_keywords = [ "success mounting image", "there is already a developer image mounted", ] if any(k in text_lower for k in success_keywords): print("[ios] 设备 %s 开发者镜像挂载完成", udid) if text.strip(): print("[ios][image auto] output:\n%s", text.strip()) return # 到这里说明没找到成功关键字，当成“不可靠但非致命” print( "[ios] 设备 %s 挂载开发者镜像可能失败(rc=%s)，输出:\n%s", udid, rc, text.strip() ) # 关键：不再 raise，直接 return，让后续 runwda 继续试 return def _run_wda(self, udid: str) -> None: # ⭐ 按你验证的命令构造参数（绝对正确） args = [ f"--udid={udid}", "runwda", f"--bundleid={WdaAppBundleId}", f"--testrunnerbundleid={WdaAppBundleId}", "--xctestconfig=yolo.xctest", # ⭐ 你亲自验证成功的值 ] rc, out, err = self._run( args, desc=f"runwda({udid})", timeout=300, check=False, ) # ======================= # 对外主流程 # ======================= def activate_ios17(self, udid: str) -> None: print("[WDA] iOS17+ 激活开始，udid=%s", udid) # 1. 启动 tunnel self._spawn_tunnel() # 2. 一直等到 pair 成功（pair 不成功就没法玩了，直接返回） try: self._pair_until_success(udid) except Exception as e: print("[WDA] pair 失败，终止激活流程 udid=%s, err=%s", udid, e) return # 3. 挂载开发者镜像（现在是非致命错误） try: self._mount_dev_image(udid) except Exception as e: # 理论上不会再进到这里，但为了稳妥，多一层保护 print("[WDA] 挂载开发者镜像出现异常，忽略继续 udid=%s, err=%s", udid, e) # 4. 尝试启动 WDA try: self._run_wda(udid) except Exception as e: print("[WDA] runwda 调用异常 udid=%s, err=%s", udid, e) print("[WDA] iOS17+ 激活流程结束（不代表一定成功），udid=%s", udid)