cmautobuy/client/src/android_device_service.py

"""通过 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