# -*- coding: utf-8 -*- import os import signal import sys import time import wda import threading import subprocess from pathlib import Path from typing import List, Dict, Optional from tidevice import Usbmux, ConnectionType from tidevice._device import BaseDevice from Entity.DeviceModel import DeviceModel from Entity.Variables import WdaAppBundleId from Module.FlaskSubprocessManager import FlaskSubprocessManager from Utils.LogManager import LogManager class Deviceinfo(object): """设备生命周期管理：以 deviceModelList 为唯一真理源""" def __init__(self): self.deviceIndex = 0 self.screenProxy = 9110 self.pidList: List[Dict] = [] # 仅记录 iproxy 进程 self.manager = FlaskSubprocessManager.get_instance() self.deviceModelList: List[DeviceModel] = [] # 根基，不动 self.maxDeviceCount = 6 self._lock = threading.Lock() self._model_index: Dict[str, DeviceModel] = {} # udid -> model self._miss_count: Dict[str, int] = {} # udid -> 连续未扫描到次数 self._port_pool: List[int] = [] # 端口回收池 self._port_in_use: set[int] = set() # 正在使用的端口 # 🔥1. 启动 WDA 健康检查线程 # threading.Thread(target=self._wda_health_checker, daemon=True).start() # region iproxy 初始化 try: self.iproxy_path = self._iproxy_path() self.iproxy_dir = self.iproxy_path.parent os.environ["PATH"] = str(self.iproxy_dir) + os.pathsep + os.environ.get("PATH", "") try: os.add_dll_directory(str(self.iproxy_dir)) except Exception: pass self._creationflags = 0x08000000 if os.name == "nt" else 0 self._popen_kwargs = dict( stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=str(self.iproxy_dir), shell=False, text=True, creationflags=self._creationflags, encoding="utf-8", bufsize=1, ) def _spawn_iproxy(udid: str, local_port: int, remote_port: int = 9100) -> subprocess.Popen: args = [str(self.iproxy_path), "-u", udid, str(local_port), str(remote_port)] p = subprocess.Popen(args, **self._popen_kwargs) def _pipe_to_log(name: str, stream): try: for line in iter(stream.readline, ''): s = line.strip() if s: LogManager.info(f"[iproxy {name}] {s}", udid) except Exception: pass threading.Thread(target=_pipe_to_log, args=("STDOUT", p.stdout), daemon=True).start() threading.Thread(target=_pipe_to_log, args=("STDERR", p.stderr), daemon=True).start() return p self._spawn_iproxy = _spawn_iproxy LogManager.info(f"iproxy 启动器已就绪，目录: {self.iproxy_dir}") except Exception as e: self.iproxy_path = None self.iproxy_dir = None self._spawn_iproxy = None LogManager.error(f"初始化 iproxy 失败：{e}") # endregion def startDeviceListener(self): while True: try: lists = Usbmux().device_list() except Exception as e: LogManager.warning(f"usbmuxd 连接失败: {e}，2 秒后重试") time.sleep(2) continue now_udids = {d.udid for d in lists if d.conn_type == ConnectionType.USB} # 1. 失踪登记 & 累加 need_remove = None # ← 新增：放锁外记录 with self._lock: for udid in list(self._model_index.keys()): if udid not in now_udids: self._miss_count[udid] = self._miss_count.get(udid, 0) + 1 if self._miss_count[udid] >= 3: self._miss_count.pop(udid, None) need_remove = udid # ← 只记录，不调用 else: self._miss_count.pop(udid, None) # 🔓 锁已释放，再删设备（不会重入） if need_remove: self._remove_model(need_remove) # 2. 全新插入（只处理未在线且信任且未满） for d in lists: if d.conn_type != ConnectionType.USB: continue udid = d.udid with self._lock: if udid in self._model_index: continue if not self.is_device_trusted(udid): continue if len(self.deviceModelList) >= self.maxDeviceCount: continue try: self.connectDevice(udid) except Exception as e: LogManager.error(f"连接设备失败 {udid}: {e}", udid) time.sleep(1) # 🔥2. WDA 健康检查 def _wda_health_checker(self): while True: time.sleep(1) print(len(self.deviceModelList)) with self._lock: online = [m for m in self.deviceModelList if m.ready] # ← 只检查就绪的 print(len(online)) for model in online: udid = model.deviceId if not self._wda_ok(udid): LogManager.warning(f"WDA 异常，重启通道：{udid}", udid) with self._lock: self._remove_model(udid) self.connectDevice(udid) # 🔥3. 真正做 health-check 的地方 def _wda_ok(self, udid: str) -> bool: """返回 True 表示 WDA 活着，False 表示已死""" try: # 用 2 秒超时快速探测 c = wda.USBClient(udid, 8100) # 下面这句就是“xctest launched but check failed” 的触发点 # 如果 status 里返回了 WebDriverAgent 运行信息就认为 OK st = c.status() if st.get("state") != "success": return False # 你也可以再苛刻一点，多做一次 /wda/healthcheck # c.http.get("/wda/healthcheck") return True except Exception as e: # 任何异常（连接拒绝、超时、json 解析失败）都认为已死 LogManager.error(f"WDA health-check 异常：{e}", udid) return False # region ===================== 增删改查唯一入口（线程安全） ===================== def _has_model(self, udid: str) -> bool: return udid in self._model_index def _add_model(self, model: DeviceModel): if model.deviceId in self._model_index: return # 防重复 model.ready = True self.deviceModelList.append(model) self._model_index[model.deviceId] = model try: self.manager.send(model.toDict()) except Exception as e: LogManager.warning(f"{model.deviceId} 发送上线事件失败：{e}") LogManager.method_info(f"{model.deviceId} 加入设备成功，当前在线数：{len(self.deviceModelList)}",method="device_count") # 删除设备 def _remove_model(self, udid: str): print(f"【删】进入删除方法 udid={udid}") LogManager.method_info(f"【删】进入删除方法 udid={udid}", method="device_count") # 1. 纯内存临界区——毫秒级 with self._lock: print(f"【删】拿到锁 udid={udid}") LogManager.method_info(f"【删】拿到锁 udid={udid}", method="device_count") model = self._model_index.pop(udid, None) if not model: print(f"【删】模型已空，直接返回 udid={udid}") LogManager.method_info(f"【删】模型已空，直接返回 udid={udid}",method="device_count") return if model.deleting: print(f"【删】正在删除中，幂等返回 udid={udid}") LogManager.method_info(method="device_count", text=f"【删】正在删除中，幂等返回 udid={udid}") return model.deleting = True # 标记维删除设备 model.type = 2 print(f"【删】标记 deleting=True udid={udid}") LogManager.method_info("【删】标记 deleting=True udid={udid}","device_count") # 过滤列表 before = len(self.deviceModelList) self.deviceModelList = [m for m in self.deviceModelList if m.deviceId != udid] after = len(self.deviceModelList) print(f"【删】列表过滤 before={before} → after={after} udid={udid}") LogManager.method_info(f"【删】列表过滤 before={before} → after={after} udid={udid}","device_count") # 端口 self._port_in_use.discard(model.screenPort) self._port_pool.append(model.screenPort) print(f"【删】回收端口 port={model.screenPort} udid={udid}") LogManager.method_info(f"【删】回收端口 port={model.screenPort} udid={udid}", method="device_count") # 进程 to_kill = [item for item in self.pidList if item.get("id") == udid] self.pidList = [item for item in self.pidList if item.get("id") != udid] print(f"【删】待杀进程数 count={len(to_kill)} udid={udid}") LogManager.method_info(f"【删】待杀进程数 count={len(to_kill)} udid={udid}", method="device_count") # 2. IO 区无锁 for idx, item in enumerate(to_kill, 1): print(f"【删】杀进程 {idx}/{len(to_kill)} pid={item.get('target').pid} udid={udid}") LogManager.method_error(f"【删】杀进程 {idx}/{len(to_kill)} pid={item.get('target').pid} udid={udid}", method="device_count") self._terminate_proc(item.get("target")) print(f"【删】进程清理完成 udid={udid}") LogManager.method_info(f"【删】进程清理完成 udid={udid}", method="device_count") # 3. 网络 IO retry = 3 while retry: try: self.manager.send(model.toDict()) print(f"【删】下线事件已发送 udid={udid}") LogManager.method_info(f"【删】下线事件已发送 udid={udid}", method="device_count") break except Exception as e: retry -= 1 print(f"【删】发送事件失败 retry={retry} err={e} udid={udid}") LogManager.method_error(f"【删】发送事件失败 retry={retry} err={e} udid={udid}", method="device_count") time.sleep(0.2) else: print(f"【删】发送事件彻底失败，主动退出 udid={udid}") LogManager.method_error(f"【删】发送事件彻底失败，主动退出 udid={udid}", method="device_count") print(f"【删】===== 设备 {udid} 删除全流程结束 =====") LogManager.method_info(f"【删】===== 设备 {udid} 删除全流程结束 =====", method="device_count") print(len(self.deviceModelList)) LogManager.method_info(f"当前剩余设备数量：{len(self.deviceModelList)}", method="device_count") # region ===================== 端口分配与回收 ===================== def _alloc_port(self) -> int: if self._port_pool: port = self._port_pool.pop() else: self.screenProxy += 1 port = self.screenProxy self._port_in_use.add(port) return port def _free_port(self, port: int): if port in self._port_in_use: self._port_in_use.remove(port) self._port_pool.append(port) # endregion # region ===================== 单台设备连接 ===================== def connectDevice(self, udid: str): if not self.is_device_trusted(udid): LogManager.warning("设备未信任，跳过 WDA 启动", udid) return try: d = wda.USBClient(udid, 8100) except Exception as e: LogManager.error(f"启动 WDA 失败: {e}", udid) return width, height, scale = 0, 0, 1.0 try: size = d.window_size() width, height = size.width, size.height scale = d.scale except Exception as e: LogManager.warning(f"读取屏幕信息失败：{e}", udid) port = self._alloc_port() model = DeviceModel(udid, port, width, height, scale, type=1) self._add_model(model) try: d.app_start(WdaAppBundleId) d.home() except Exception as e: LogManager.warning(f"启动/切回桌面失败：{e}", udid) time.sleep(2) # 先清旧进程再启动新进程 self.pidList = [item for item in self.pidList if item.get("id") != udid] target = self.relayDeviceScreenPort(udid, port) if target: self.pidList.append({"target": target, "id": udid}) # region ===================== 工具方法 ===================== def is_device_trusted(self, udid: str) -> bool: try: d = BaseDevice(udid) d.get_value("DeviceName") return True except Exception: return False def relayDeviceScreenPort(self, udid: str, port: int) -> Optional[subprocess.Popen]: """启动 iproxy 前：端口若仍被占用则先杀掉占用者，再启动""" if not self._spawn_iproxy: LogManager.error("iproxy 启动器未就绪", udid) return None # --- 新增：端口冲突检查 + 强制清理 --- while self._port_in_use and self._is_port_open(port): # 先查是哪个进程占用 pid = self._get_pid_by_port(port) if pid and pid != os.getpid(): LogManager.warning(f"端口 {port} 仍被 PID {pid} 占用，尝试释放", udid) self._kill_pid_gracefully(pid) else: break # ------------------------------------- try: p = self._spawn_iproxy(udid, port, 9100) self._port_in_use.add(port) LogManager.info(f"启动 iproxy 成功，本地 {port} -> 设备 9100", udid) return p except Exception as e: LogManager.error(f"启动 iproxy 失败：{e}", udid) return None # ------------------- 新增三个小工具 ------------------- def _is_port_open(self, port: int) -> bool: import socket with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: return s.connect_ex(("127.0.0.1", port)) == 0 def _get_pid_by_port(self, port: int) -> Optional[int]: """跨平台根据端口号查 PID，失败返回 None""" try: if os.name == "nt": cmd = ["netstat", "-ano", "-p", "tcp"] out = subprocess.check_output(cmd, text=True) for line in out.splitlines(): if f"127.0.0.1:{port}" in line and "LISTENING" in line: return int(line.strip().split()[-1]) else: cmd = ["lsof", "-t", f"-iTCP:{port}", "-sTCP:LISTEN"] out = subprocess.check_output(cmd, text=True) return int(out.strip().split()[0]) except Exception: return None def _kill_pid_gracefully(self, pid: int): """先 terminate 再 kill -9""" try: os.kill(pid, signal.SIGTERM) time.sleep(1) os.kill(pid, signal.SIGKILL) except Exception: pass def _terminate_proc(self, p: Optional[subprocess.Popen]): if not p or p.poll() is not None: return try: p.terminate() p.wait(timeout=3) except Exception: try: if os.name == "posix": os.killpg(os.getpgid(p.pid), signal.SIGKILL) else: p.kill() p.wait(timeout=2) except Exception: pass def _base_dir(self) -> Path: if getattr(sys, "frozen", False): return Path(sys.executable).resolve().parent return Path(__file__).resolve().parents[1] def _iproxy_path(self) -> Path: exe = "iproxy.exe" if os.name == "nt" else "iproxy" base = self._base_dir() candidates = [base / "resources" / "iproxy" / exe] for p in candidates: if p.exists(): return p raise FileNotFoundError(f"iproxy not found, tried: {[str(c) for c in candidates]}")