371 lines
13 KiB
Python
371 lines
13 KiB
Python
"""通过 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_USB_DISCONNECT_TIMEOUT_SECONDS = 60.0
|
||
DEFAULT_USB_POLL_INTERVAL_SECONDS = 0.5
|
||
|
||
|
||
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]
|
||
Sleeper = Callable[[float], None]
|
||
ProgressCallback = Callable[[str], 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,
|
||
usb_disconnect_timeout_seconds: float = (
|
||
DEFAULT_USB_DISCONNECT_TIMEOUT_SECONDS
|
||
),
|
||
usb_poll_interval_seconds: float = DEFAULT_USB_POLL_INTERVAL_SECONDS,
|
||
sleeper: Sleeper = time.sleep,
|
||
):
|
||
if timeout_seconds <= 0:
|
||
raise ValueError("ADB 超时时间必须大于 0")
|
||
if usb_disconnect_timeout_seconds <= 0:
|
||
raise ValueError("等待拔出 USB 的超时时间必须大于 0")
|
||
if usb_poll_interval_seconds <= 0:
|
||
raise ValueError("USB 检查间隔必须大于 0")
|
||
self._run_command = command_runner
|
||
self._timeout_seconds = timeout_seconds
|
||
self._usb_disconnect_timeout_seconds = usb_disconnect_timeout_seconds
|
||
self._usb_poll_interval_seconds = usb_poll_interval_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 convert_usb_to_wifi(
|
||
self,
|
||
usb_serial: str,
|
||
is_cancelled: Optional[Callable[[], bool]] = None,
|
||
on_progress: Optional[ProgressCallback] = None,
|
||
) -> AndroidWifiConversionResult:
|
||
"""开启 USB 设备的 5555 端口，等拔线后重连并返回最新列表。"""
|
||
|
||
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 调试",
|
||
)
|
||
|
||
# 部分手机插线时就能建立 Wi-Fi 通道；该连接仅用于提前准备，
|
||
# 真正的结果仍以拔线后再次连接和验证为准。
|
||
try:
|
||
self._connect_wifi(wifi_serial)
|
||
except AndroidDeviceSearchError:
|
||
pass
|
||
|
||
self._report(
|
||
on_progress,
|
||
f"已开启 {wifi_serial}，请拔掉 USB 数据线…",
|
||
)
|
||
self._wait_until_usb_disconnected(usb_serial, is_cancelled)
|
||
|
||
self._check_cancelled(is_cancelled)
|
||
self._report(on_progress, "已检测到 USB 拔出，正在重新连接 Wi-Fi…")
|
||
self._connect_wifi(wifi_serial)
|
||
|
||
self._check_cancelled(is_cancelled)
|
||
devices = self._list_devices()
|
||
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 已连接，正在刷新设备列表…")
|
||
return AndroidWifiConversionResult(wifi_serial, self.search(is_cancelled))
|
||
|
||
@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_usb_disconnected(
|
||
self,
|
||
usb_serial: str,
|
||
is_cancelled: Optional[Callable[[], bool]],
|
||
) -> None:
|
||
attempts = max(
|
||
1,
|
||
int(
|
||
self._usb_disconnect_timeout_seconds
|
||
/ self._usb_poll_interval_seconds
|
||
),
|
||
)
|
||
missing_checks = 0
|
||
for attempt in range(attempts):
|
||
self._check_cancelled(is_cancelled)
|
||
devices = self._list_devices()
|
||
if all(device.serial != usb_serial for device in devices):
|
||
missing_checks += 1
|
||
if missing_checks >= 2:
|
||
return
|
||
else:
|
||
missing_checks = 0
|
||
if attempt < attempts - 1:
|
||
self._sleep(self._usb_poll_interval_seconds)
|
||
|
||
raise AndroidDeviceSearchError(
|
||
"等待拔出 USB 超时；5555 已开启，请拔线后重新点击转换"
|
||
)
|
||
|
||
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 _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
|