iOSAI/Module/DeviceInfo.py

# -*- coding: utf-8 -*-
import os
import sys
import time
import json
import wda
import threading
import subprocess
from pathlib import Path
from typing import List, Dict, Optional

from tidevice import Usbmux
from Entity.DeviceModel import DeviceModel
from Entity.Variables import WdaAppBundleId
from Module.FlaskSubprocessManager import FlaskSubprocessManager
from Utils.LogManager import LogManager


class Deviceinfo(object):
    def __init__(self):
        self.deviceIndex = 0
        # 投屏端口（本地映射端口起始值，会递增）
        self.screenProxy = 9110

        # 记录 iproxy Popen 进程：[{ "id": udid, "target": Popen }, ...]
        self.pidList: List[Dict] = []
        # 当前已连接的设备（tidevice 的 Device 对象列表）
        self.deviceArray: List = []

        # 子进程通信（向前端发送设备信息）
        self.manager = FlaskSubprocessManager.get_instance()
        # 已发给前端的设备模型列表（用于拔出时发 type=2）
        self.deviceModelList: List[DeviceModel] = []

    # ----------------------------
    # 监听设备连接（死循环，内部捕获异常）
    # ----------------------------
    def startDeviceListener(self):
        LogManager.info("Device Listener started", "listener")
        while True:
            try:
                lists = Usbmux().device_list()
            except Exception as e:
                # 另一台电脑常见：usbmuxd 连接失败（未安装 iTunes/Apple Mobile Device Support）
                LogManager.warning(f"usbmuxd 连接失败: {e}。请确认已安装 iTunes/Apple Mobile Device Support，并在手机上“信任此电脑”", "listener")
                time.sleep(2)
                continue

            # 新接入设备
            for device in lists:
                if device not in self.deviceArray:
                    self.screenProxy += 1
                    try:
                        self.connectDevice(device.udid)
                        self.deviceArray.append(device)
                    except Exception as e:
                        LogManager.error(f"连接设备失败 {device.udid}: {e}", device.udid)

            # 拔出设备处理
            self._removeDisconnected(lists)

            time.sleep(1)

    # ----------------------------
    # 连接单台设备：启动 WDA、读取屏参、通知前端、映射投屏端口
    # ----------------------------
    def connectDevice(self, identifier: str):
        # 1) 连接 WDA（USBClient -> 设备 8100）
        try:
            d = wda.USBClient(identifier, 8100)
            LogManager.info("启动 WDA 成功", identifier)
        except Exception as e:
            LogManager.error(f"启动 WDA 失败，请检查手机是否已信任、WDA 是否正常。错误: {e}", identifier)
            return  # 不抛出到外层，保持监听循环健壮

        # 2) 读取屏幕信息（失败不影响主流程）
        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}", identifier)

        # 3) 组装模型并发送给前端
        model = DeviceModel(identifier, self.screenProxy, width, height, scale, type=1)
        self.deviceModelList.append(model)
        try:
            self.manager.send(model.toDict())
        except Exception as e:
            LogManager.warning(f"向前端发送设备模型失败：{e}", identifier)

        # 4) 可选：启动你的 app 并回到桌面
        try:
            d.app_start(WdaAppBundleId)
            d.home()
        except Exception as e:
            LogManager.warning(f"启动/切回桌面失败：{e}", identifier)

        time.sleep(2)

        # 5) 本地端口 -> 设备端口 的映射（投屏：本地 self.screenProxy -> 设备 9100）
        target = self.relayDeviceScreenPort(identifier)
        self.pidList.append({"target": target, "id": identifier})

    # ----------------------------
    # 处理拔出设备：发通知、关掉 iproxy、移出状态
    # ----------------------------
    def _removeDisconnected(self, current_list):
        set1 = set(self.deviceArray)
        set2 = set(current_list)
        difference = list(set1 - set2)  # 在旧集合中但不在新集合中 -> 已拔出

        for i in difference:
            udid = i.udid
            # 1) 通知前端：type = 2
            for a in list(self.deviceModelList):
                if udid == a.deviceId:
                    a.type = 2
                    try:
                        self.manager.send(a.toDict())
                    except Exception as e:
                        LogManager.warning(f"发送下线事件失败：{e}", udid)
                    self.deviceModelList.remove(a)

            # 2) 关掉对应的 iproxy
            for k in list(self.pidList):
                if udid == k["id"]:
                    target = k.get("target")
                    try:
                        if target and target.poll() is None:
                            target.kill()
                    except Exception:
                        pass
                    self.pidList.remove(k)

            # 3) 从已连接集合中移除
            try:
                self.deviceArray.remove(i)
            except Exception:
                pass


    # ----------------------------
    # 路径：打包/源码都能找到根目录、iproxy 目录和 iproxy 可执行文件
    # ----------------------------
    def _base_dir(self) -> Path:
        """
        打包后：返回 exe 所在目录；
        源码运行：返回项目根目录（Module 的上一级）
        """
        if getattr(sys, "frozen", False):
            return Path(sys.executable).resolve().parent
        return Path(__file__).resolve().parents[1]   # iOSAI/ 作为根

    def _iproxy_dir(self) -> Path:
        """返回打包后的 iproxy 目录（你现在放在 Module/iproxy/）"""
        return self._base_dir() / "Module" / "iproxy"

    def _iproxy_path(self) -> Path:
        """返回 iproxy 可执行文件路径（Windows 为 iproxy.exe）"""
        exe_name = "iproxy.exe" if os.name == "nt" else "iproxy"
        return self._iproxy_dir() / exe_name

    # ----------------------------
    # 端口映射：启动 iproxy（设置 cwd 和 PATH，隐藏窗口）
    # ----------------------------
    def relayDeviceScreenPort(self, udid: str) -> Optional[subprocess.Popen]:
        try:
            iproxy = self._iproxy_path()
            iproxy_dir = self._iproxy_dir()

            if not iproxy.exists():
                raise FileNotFoundError(f"iproxy not found: {iproxy}")

            # 继承环境并把 iproxy 目录加入 PATH，方便 DLL 解析
            env = os.environ.copy()
            env["PATH"] = str(iproxy_dir) + os.pathsep + env.get("PATH", "")

            # Windows 隐藏子进程窗口
            CREATE_NO_WINDOW = 0x08000000 if os.name == "nt" else 0

            p = subprocess.Popen(
                [str(iproxy), "-u", udid, str(self.screenProxy), "9100"],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                creationflags=CREATE_NO_WINDOW,
                cwd=str(iproxy_dir),   # 关键：工作目录设为 iproxy 所在目录
                env=env,               # 关键：把 iproxy_dir 注入 PATH
                text=True,             # 你后面如果要读 stdout/stderr 的话更方便
                encoding="utf-8",
                bufsize=1
            )
            LogManager.info(f"启动 iproxy 成功，本地 {self.screenProxy} -> 设备 9100", udid)
            return p

        except Exception as e:
            LogManager.error(f"启动 iproxy 失败：{e}", udid)
            return None