feat: 实现 Android 设备搜索 (#21)

client/src/android_device_service.py

@@ -0,0 +1,159 @@
+"""通过 ADB 查询当前电脑已经识别的 Android 设备。"""
+
+from dataclasses import dataclass
+import subprocess
+from typing import Callable, List, Optional, Sequence
+
+
+DEFAULT_ADB_TIMEOUT_SECONDS = 5.0
+
+
+class AndroidDeviceSearchError(RuntimeError):
+    """ADB 无法完成设备搜索时抛出的可读错误。"""
+
+
+@dataclass(frozen=True)
+class AndroidDevice:
+    """一台由 ``adb devices -l`` 返回的设备。"""
+
+    serial: str
+    connection_type: str
+    model: str = "—"
+    android_version: str = "—"
+    status: str = "device"
+
+
+CommandRunner = Callable[[Sequence[str], float], subprocess.CompletedProcess]
+
+
+def run_adb_command(
+    command: Sequence[str], timeout_seconds: float
+) -> subprocess.CompletedProcess:
+    """执行一条隐藏窗口的 ADB 命令，返回标准输出和错误输出。"""
+
+    return subprocess.run(
+        list(command),
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        timeout=timeout_seconds,
+        check=False,
+        creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
+    )
+
+
+class AndroidDeviceService:
+    """查询 ADB 设备列表；本类不依赖 Qt，也不创建 uiautomator2 设备。"""
+
+    def __init__(
+        self,
+        command_runner: CommandRunner = run_adb_command,
+        timeout_seconds: float = DEFAULT_ADB_TIMEOUT_SECONDS,
+    ):
+        if timeout_seconds <= 0:
+            raise ValueError("ADB 超时时间必须大于 0")
+        self._run_command = command_runner
+        self._timeout_seconds = timeout_seconds
+
+    def search(
+        self, is_cancelled: Optional[Callable[[], bool]] = None
+    ) -> List[AndroidDevice]:
+        """返回当前 ADB 设备；取消后停止补充设备属性。"""
+
+        result = self._run(["adb", "devices", "-l"], "搜索设备")
+        devices = self.parse_devices(result.stdout or "")
+
+        enriched = []
+        for device in devices:
+            if is_cancelled is not None and is_cancelled():
+                break
+            if device.status != "device":
+                enriched.append(device)
+                continue
+
+            model = device.model
+            if model == "—":
+                model = self._read_property(device.serial, "ro.product.model")
+            if is_cancelled is not None and is_cancelled():
+                break
+            android_version = self._read_property(
+                device.serial, "ro.build.version.release"
+            )
+            enriched.append(
+                AndroidDevice(
+                    serial=device.serial,
+                    connection_type=device.connection_type,
+                    model=model,
+                    android_version=android_version,
+                    status=device.status,
+                )
+            )
+        return enriched
+
+    @staticmethod
+    def parse_devices(output: str) -> List[AndroidDevice]:
+        """解析 ``adb devices -l`` 输出，不依赖操作系统换行格式。"""
+
+        devices = []
+        for raw_line in output.splitlines():
+            line = raw_line.strip()
+            if not line or line.startswith("List of devices") or line.startswith("*"):
+                continue
+
+            parts = line.split()
+            if len(parts) < 2:
+                continue
+
+            serial, status = parts[0], parts[1]
+            fields = {}
+            for token in parts[2:]:
+                if ":" not in token:
+                    continue
+                key, value = token.split(":", 1)
+                fields[key] = value
+
+            model = fields.get("model", "—").replace("_", " ") or "—"
+            devices.append(
+                AndroidDevice(
+                    serial=serial,
+                    connection_type="Wi-Fi" if ":" in serial else "USB",
+                    model=model,
+                    status=status,
+                )
+            )
+        return devices
+
+    def _read_property(self, serial: str, property_name: str) -> str:
+        """读取单个属性；失败只返回占位符，不丢掉已经发现的设备。"""
+
+        try:
+            result = self._run(
+                ["adb", "-s", serial, "shell", "getprop", property_name],
+                "读取设备信息",
+            )
+        except AndroidDeviceSearchError:
+            return "—"
+        value = (result.stdout or "").strip()
+        return value or "—"
+
+    def _run(self, command: Sequence[str], action: str):
+        try:
+            result = self._run_command(command, self._timeout_seconds)
+        except FileNotFoundError as exc:
+            raise AndroidDeviceSearchError(
+                "未找到 adb，请安装 Android platform-tools 并把 adb 加入 PATH 后重试"
+            ) from exc
+        except subprocess.TimeoutExpired as exc:
+            raise AndroidDeviceSearchError(
+                f"ADB {action}超时，请检查设备连接或重启 ADB 后重试"
+            ) from exc
+        except OSError as exc:
+            raise AndroidDeviceSearchError(
+                f"无法启动 adb：{str(exc) or '系统拒绝执行'}"
+            ) from exc
+
+        if result.returncode != 0:
+            detail = (result.stderr or result.stdout or "未知错误").strip()
+            raise AndroidDeviceSearchError(f"ADB {action}失败：{detail}")
+        return result