diff --git a/Module/DeviceInfo.py b/Module/DeviceInfo.py index 08e93bf..16a56dc 100644 --- a/Module/DeviceInfo.py +++ b/Module/DeviceInfo.py @@ -69,33 +69,53 @@ class DeviceInfo: LogManager.method_info("进入主循环", "listen", udid="system") print("[Listen] 开始监听设备上下线...") - last_broadcast = 0.0 # 用来做“心跳全量同步”的时间戳 - while True: try: usb = Usbmux().device_list() + # 只看 USB 连接的设备 online = {d.udid for d in usb if d.conn_type == ConnectionType.USB} except Exception as e: LogManager.warning(f"[device_list] 异常：{e}", udid="system") - print("[Listen] device_list 出错，本轮视为无设备，防止状态脏死") - usb = [] - online = set() + time.sleep(1) + continue + now = time.time() + + # 当前已知的设备（本轮循环开始时） with self._lock: known = set(self._models.keys()) - # ---------- 1. 新设备 ---------- - now = time.time() + # 1. 处理在线设备 for udid in online: + # 更新心跳时间 self._last_seen[udid] = now - if udid not in known: - try: - self._add_device(udid) - except Exception as e: - LogManager.warning(f"[Add] 处理设备 {udid} 异常: {e}", udid=udid) - print(f"[Add] 处理设备 {udid} 异常: {e}") - # ---------- 2. 可能离线设备 ---------- + # 已经在列表里的设备，直接跳过“是否信任”检查 + if udid in known: + continue + + # 只对“新发现”的设备做一次信任检查 + try: + if not self._is_trusted(udid): + # 未信任 / 未配对 / 暂时不可用，本轮直接跳过 + LogManager.info(f"[Add] 设备未信任或未就绪，跳过本轮添加: {udid}", udid=udid) + print(f"[Add] 设备未信任或未就绪，跳过: {udid}") + continue + except Exception as e: + # 信任检查本身异常，也当作暂时未就绪处理 + LogManager.warning(f"[Add] 检测设备 {udid} 信任状态异常: {e}", udid=udid) + print(f"[Add] 检测设备 {udid} 信任状态异常: {e}") + continue + + # 确认“已信任”的新设备，才真正走 _add_device + try: + self._add_device(udid) + except Exception as e: + # 单设备异常不能干掉整个循环 + LogManager.warning(f"[Add] 处理设备 {udid} 异常: {e}", udid=udid) + print(f"[Add] 处理设备 {udid} 异常: {e}") + + # 2. 处理可能离线的设备（只看本轮开始时 known 里的） for udid in list(known): if udid not in online: last = self._last_seen.get(udid, 0) @@ -106,17 +126,26 @@ class DeviceInfo: LogManager.method_error(f"移除失败：{e}", "listen", udid=udid) print(f"[Remove] 移除失败 {udid}: {e}") - # ---------- 3. 心跳：每 5 秒强制同步一次到 Flask ---------- - if now - last_broadcast > 5.0: - try: - self._manager_send() - except Exception as e: - print(f"[Listen] 周期同步到 Flask 失败: {e}") - else: - last_broadcast = now - time.sleep(1) + # 判断设备是否信任 + def _is_trusted(self, udid: str) -> bool: + try: + d = tidevice.Device(udid) + # 随便读一个需要 lockdown/配对的字段 + _ = d.product_version # 或 d.info，视你当前 tidevice 版本而定 + return True + except Exception as e: + msg = str(e) + # 这里只是示意，你可以根据你本地真实报错关键字再精细一点 + if "NotTrusted" in msg or "Please trust" in msg or "InvalidHostID" in msg: + print(f"[Trust] 设备未信任，udid={udid}, err={msg}") + return False + + # 如果是别的错误（比如瞬时通信异常），我倾向于当作“暂时不信任”，避免把有问题的设备加进去 + print(f"[Trust] 检测信任状态出错，当作未信任处理 udid={udid}, err={msg}") + return False + # ========================== # 添加设备 # ========================== diff --git a/Module/IOSActivator.py b/Module/IOSActivator.py index 1f0e15b..50a8a2b 100644 --- a/Module/IOSActivator.py +++ b/Module/IOSActivator.py @@ -4,13 +4,15 @@ import time import threading import subprocess from typing import Optional, Callable + from Entity.Variables import WdaAppBundleId + class IOSActivator: """ 给 iOS17+ 用的 go-ios 激活器（单例）： - 维护一条全局 tunnel 进程 - - 流程：tunnel start -> pair(重试) -> image auto(不致命) -> runwda(多次重试+日志判定成功) + - 流程：tunnel start -> pair(可多次重试) -> image auto(非致命) -> runwda(多次重试+日志判定成功) - WDA 启动成功后触发回调 on_wda_ready(udid) """ @@ -28,13 +30,13 @@ class IOSActivator: return cls._instance def __init__( - self, - ios_path: Optional[str] = None, - pair_timeout: int = 60, # 配对最多等多久 - pair_retry_interval: int = 3, # 配对重试间隔 - runwda_max_retry: int = 10, # runwda 最大重试次数 - runwda_retry_interval: int = 3,# runwda 重试间隔 - runwda_wait_timeout: int = 25 # 单次 runwda 等待“成功日志”的超时时间 + self, + ios_path: Optional[str] = None, + pair_timeout: int = 60, # 配对最多等多久 + pair_retry_interval: int = 3, # 配对重试间隔 + runwda_max_retry: int = 10, # runwda 最大重试次数 + runwda_retry_interval: int = 3, # runwda 重试间隔 + runwda_wait_timeout: int = 25 # 单次 runwda 等待“成功日志”的超时时间 ): if getattr(self, "_inited", False): return @@ -59,18 +61,25 @@ class IOSActivator: self._lock = threading.Lock() - # Windows 隐藏黑框 + # ========= 关键：这里改成“真正隐藏窗口”的安全版 ========= self._creationflags = 0 self._startupinfo = None if os.name == "nt": try: + # 只用 CREATE_NO_WINDOW，不搞 DETACHED_PROCESS / NEW_PROCESS_GROUP 之类的骚操作 self._creationflags = subprocess.CREATE_NO_WINDOW # type: ignore[attr-defined] except Exception: self._creationflags = 0 + si = subprocess.STARTUPINFO() - si.dwFlags |= subprocess.STARTF_USESHOWWINDOW # type: ignore[attr-defined] + try: + si.dwFlags |= subprocess.STARTF_USESHOWWINDOW # type: ignore[attr-defined] + except Exception: + # 某些极端环境下可能没有 STARTF_USESHOWWINDOW，忽略即可 + pass si.wShowWindow = 0 # SW_HIDE self._startupinfo = si + # ========= 关键部分结束 ========= self._inited = True @@ -134,8 +143,9 @@ class IOSActivator: print(f"[ios][{name}] 读取 stderr 异常: {e}") def _spawn_tunnel(self): - """启动 / 复用全局 tunnel""" + """启动 / 复用全局 tunnel（不隐藏窗口）""" with IOSActivator._tunnel_lock: + # 已有并且还在跑就复用 if IOSActivator._tunnel_proc is not None and IOSActivator._tunnel_proc.poll() is None: print("[ios] tunnel 已经在运行，跳过重新启动") return @@ -148,8 +158,8 @@ class IOSActivator: stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, - creationflags=self._creationflags, - startupinfo=self._startupinfo, + creationflags=self._creationflags, # 0：不隐藏 + startupinfo=self._startupinfo, # None：不隐藏 ) except Exception as e: print("[ios] 启动 tunnel 失败(忽略):", e) @@ -158,6 +168,7 @@ class IOSActivator: IOSActivator._tunnel_proc = proc print("[ios] tunnel 启动成功, PID=", proc.pid) + # 后台吃日志 threading.Thread( target=self._drain_process_output, args=(proc, "tunnel"), @@ -181,6 +192,10 @@ class IOSActivator: ) text = (out or "") + "\n" + (err or "") + # 打印一份完整输出，方便调试 + if text.strip(): + print("[ios][pair] output:\n", text.strip()) + if "Successfully paired" in text: print(f"[ios] 设备 {udid} 配对成功") return @@ -238,7 +253,7 @@ class IOSActivator: stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, - creationflags=self._creationflags, + creationflags=self._creationflags, # 0：不隐藏 startupinfo=self._startupinfo, ) except Exception as e: @@ -255,7 +270,7 @@ class IOSActivator: continue print(f"[WDA-LOG] {line}") lower = line.lower() - # 这里是你实测的“成功特征” + # 你实测的“成功特征” if "got capabilities" in lower or '"authorized":true' in lower: success_evt.set() print(f"[ios] 捕获到 WDA 启动成功日志({tag})，udid={udid}") @@ -316,7 +331,7 @@ class IOSActivator: def activate_ios17(self, udid: str, on_wda_ready: Optional[Callable[[str], None]] = None) -> None: print(f"[WDA] iOS17+ 激活开始，udid={udid}, 回调={on_wda_ready}") - # 1. tunnel + # 1. 先确保 tunnel 在跑 self._spawn_tunnel() # 2. 配对