cmautobuy/client/src/android_device_service.py

"""通过 ADB 查询当前电脑已经识别的 Android 设备。"""

from dataclasses import dataclass
import ipaddress
import subprocess
import time
from typing import Callable, List, Optional, Sequence


DEFAULT_ADB_TIMEOUT_SECONDS = 5.0
DEFAULT_DEVICE_READY_CHECKS = 8
DEFAULT_DEVICE_READY_INTERVAL_SECONDS = 0.25


class AndroidDeviceSearchError(RuntimeError):
    """ADB 无法完成设备搜索时抛出的可读错误。"""


class AndroidDeviceConversionCancelled(RuntimeError):
    """页面关闭后停止 USB 转 Wi-Fi 的后续步骤。"""


@dataclass(frozen=True)
class AndroidDevice:
    """一台由 ``adb devices -l`` 返回的设备。"""

    serial: str
    connection_type: str
    model: str = "—"
    android_version: str = "—"
    status: str = "device"


@dataclass(frozen=True)
class AndroidWifiConversionResult:
    """USB 转 Wi-Fi 成功后的目标设备号和最新设备列表。"""

    wifi_serial: str
    devices: List[AndroidDevice]


CommandRunner = Callable[[Sequence[str], float], subprocess.CompletedProcess]
ProgressCallback = Callable[[str], None]
Sleeper = Callable[[float], None]


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,
        sleeper: Sleeper = time.sleep,
    ):
        if timeout_seconds <= 0:
            raise ValueError("ADB 超时时间必须大于 0")
        self._run_command = command_runner
        self._timeout_seconds = timeout_seconds
        self._sleep = sleeper

    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

    def restore_saved_device(
        self,
        serial: str,
        is_cancelled: Optional[Callable[[], bool]] = None,
    ) -> List[AndroidDevice]:
        """恢复已保存设备；Wi-Fi 先重连，USB 只刷新设备列表。"""

        self._validate_saved_serial(serial)
        self._check_cancelled(is_cancelled)
        if ":" in serial:
            self._connect_wifi(serial)
            self._wait_until_device_ready(serial, is_cancelled)
        self._check_cancelled(is_cancelled)
        return self.search(is_cancelled)

    def convert_usb_to_wifi(
        self,
        usb_serial: str,
        is_cancelled: Optional[Callable[[], bool]] = None,
        on_progress: Optional[ProgressCallback] = None,
    ) -> AndroidWifiConversionResult:
        """插着数据线开启 5555，连接 Wi-Fi 后返回最新设备列表。"""

        self._validate_usb_serial(usb_serial)
        self._check_cancelled(is_cancelled)
        self._report(on_progress, "正在读取勾选设备的 Wi-Fi 地址…")
        route = self._run(
            ["adb", "-s", usb_serial, "shell", "ip", "route"],
            "读取 Wi-Fi 地址",
        )
        ip_address = self.parse_wifi_ipv4(route.stdout or "")
        wifi_serial = f"{ip_address}:5555"

        self._check_cancelled(is_cancelled)
        self._report(on_progress, "正在开启 ADB 5555 端口…")
        self._run(
            ["adb", "-s", usb_serial, "tcpip", "5555"],
            "开启 Wi-Fi 调试",
        )

        self._check_cancelled(is_cancelled)
        self._report(on_progress, f"正在连接 Wi-Fi 设备 {wifi_serial}…")
        self._connect_wifi(wifi_serial)
        self._wait_until_device_ready(wifi_serial, is_cancelled)
        self._wait_until_device_ready(usb_serial, is_cancelled)

        self._check_cancelled(is_cancelled)
        devices = self.search(is_cancelled)
        wifi_device = next(
            (device for device in devices if device.serial == wifi_serial),
            None,
        )
        if wifi_device is None or wifi_device.status != "device":
            status = wifi_device.status if wifi_device is not None else "未发现"
            raise AndroidDeviceSearchError(
                f"Wi-Fi 连接验证失败：{wifi_serial} 状态为 {status}；"
                "请确认手机和电脑在同一网络后重试"
            )

        self._report(on_progress, "Wi-Fi ADB 已连接，可以保存后拔掉 USB")
        return AndroidWifiConversionResult(wifi_serial, devices)

    @staticmethod
    def parse_wifi_ipv4(route_output: str) -> str:
        """从 Android ``ip route`` 输出的 ``src`` 字段读取有效 IPv4。"""

        route_tokens = [line.split() for line in route_output.splitlines()]
        default_interfaces = []
        for tokens in route_tokens:
            if not tokens or tokens[0] != "default" or "dev" not in tokens:
                continue
            device_index = tokens.index("dev")
            if device_index + 1 < len(tokens):
                default_interfaces.append(tokens[device_index + 1])

        preferred_routes = []
        for interface in default_interfaces:
            preferred_routes.extend(
                tokens
                for tokens in route_tokens
                if "dev" in tokens
                and tokens.index("dev") + 1 < len(tokens)
                and tokens[tokens.index("dev") + 1] == interface
            )
        preferred_routes.extend(
            tokens for tokens in route_tokens if tokens not in preferred_routes
        )

        for tokens in preferred_routes:
            for index, token in enumerate(tokens[:-1]):
                if token != "src":
                    continue
                candidate = tokens[index + 1]
                try:
                    address = ipaddress.ip_address(candidate)
                except ValueError:
                    continue
                if (
                    address.version == 4
                    and not address.is_loopback
                    and not address.is_unspecified
                    and not address.is_multicast
                ):
                    return str(address)
        raise AndroidDeviceSearchError(
            "无法读取手机 Wi-Fi IPv4；请先让手机连接与电脑相同的 Wi-Fi"
        )

    @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 _wait_until_device_ready(
        self,
        serial: str,
        is_cancelled: Optional[Callable[[], bool]],
    ) -> None:
        """短暂等待 ADB transport 完成重启或授权握手。"""

        for attempt in range(DEFAULT_DEVICE_READY_CHECKS):
            self._check_cancelled(is_cancelled)
            device = next(
                (
                    item
                    for item in self._list_devices()
                    if item.serial == serial
                ),
                None,
            )
            if device is not None and device.status == "device":
                return
            if attempt < DEFAULT_DEVICE_READY_CHECKS - 1:
                self._sleep(DEFAULT_DEVICE_READY_INTERVAL_SECONDS)

    def _list_devices(self) -> List[AndroidDevice]:
        result = self._run(["adb", "devices", "-l"], "读取设备列表")
        return self.parse_devices(result.stdout or "")

    def _connect_wifi(self, wifi_serial: str) -> None:
        result = self._run(
            ["adb", "connect", wifi_serial],
            "连接 Wi-Fi 设备",
        )
        detail = " ".join(
            part.strip() for part in (result.stdout, result.stderr) if part
        )
        normalized_detail = detail.lower()
        if (
            "connected to" not in normalized_detail
            and "already connected to" not in normalized_detail
        ):
            raise AndroidDeviceSearchError(
                f"ADB 连接 {wifi_serial} 失败："
                f"{detail or '没有返回成功信息'}"
            )

    @staticmethod
    def _validate_usb_serial(serial: str) -> None:
        if not isinstance(serial, str) or not serial.strip():
            raise AndroidDeviceSearchError("请先勾选一台已连接的 USB 设备")
        contains_whitespace = any(
            character.isspace() for character in serial
        )
        if serial != serial.strip() or contains_whitespace:
            raise AndroidDeviceSearchError("USB 设备号格式不正确")
        if ":" in serial:
            raise AndroidDeviceSearchError("当前设备已经是 Wi-Fi 连接")

    @staticmethod
    def _validate_saved_serial(serial: str) -> None:
        if not isinstance(serial, str) or not serial.strip():
            raise AndroidDeviceSearchError("已保存 Android 设备号为空")
        contains_whitespace = any(
            character.isspace() for character in serial
        )
        if serial != serial.strip() or contains_whitespace:
            raise AndroidDeviceSearchError("已保存 Android 设备号格式不正确")

    @staticmethod
    def _check_cancelled(
        is_cancelled: Optional[Callable[[], bool]],
    ) -> None:
        if is_cancelled is not None and is_cancelled():
            raise AndroidDeviceConversionCancelled()

    @staticmethod
    def _report(
        callback: Optional[ProgressCallback], message: str
    ) -> None:
        if callback is not None:
            callback(message)

    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