chengma/cmautobuy
2
0
Fork 0
Code Issues 89 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 内置并统一使用项目 ADB (#149)

Browse Source
This commit is contained in:
chengma
2026-08-11 12:01:07 +08:00
parent 56b1a46c5c
commit 8f2037db2f
21 changed files with 355 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/adb_runtime.py
+55
View File
@@ -0,0 +1,55 @@
"""项目内置 ADB 的路径、完整性检查和 adbutils 环境配置。"""
import os
from pathlib import Path
from typing import Tuple
from .db import app_dir
ADB_RELATIVE_DIRECTORY = Path("vendor/android-platform-tools/windows")
REQUIRED_ADB_FILES: Tuple[str, ...] = (
"adb.exe",
"AdbWinApi.dll",
"AdbWinUsbApi.dll",
)
class BundledAdbError(RuntimeError):
"""项目内置 ADB 缺失或不完整。"""
def bundled_adb_directory() -> Path:
"""返回项目内置 ADB 目录，不创建目录。"""
return app_dir() / ADB_RELATIVE_DIRECTORY
def validate_bundled_adb() -> Path:
"""检查 ADB 和两个配套 DLL，完整时返回 adb.exe 绝对路径。"""
directory = bundled_adb_directory()
missing_files = [
name for name in REQUIRED_ADB_FILES
if not (directory / name).is_file()
]
if missing_files:
names = "、".join(missing_files)
raise BundledAdbError(
f"内置 ADB 文件缺失：{names}。请重新安装完整的软件包。"
)
return (directory / "adb.exe").resolve()
def bundled_adb_path() -> Path:
"""返回预期的项目内 adb.exe 路径；不访问系统 PATH。"""
return (bundled_adb_directory() / "adb.exe").resolve()
def configure_bundled_adb_environment() -> Path:
"""校验内置 ADB，并强制 adbutils/uiautomator2 使用同一文件。"""
adb_path = validate_bundled_adb()
os.environ["ADBUTILS_ADB_PATH"] = str(adb_path)
return adb_path
client/src/android_device_service.py
+43 -9
View File
@@ -4,8 +4,15 @@ from dataclasses import dataclass
import ipaddress
import subprocess
import time
from pathlib import Path
from typing import Callable, List, Optional, Sequence
from .adb_runtime import (
BundledAdbError,
bundled_adb_path,
validate_bundled_adb,
)
DEFAULT_ADB_TIMEOUT_SECONDS = 5.0
DEFAULT_DEVICE_READY_CHECKS = 8
@@ -78,19 +85,24 @@ class AndroidDeviceService:
command_runner: CommandRunner = run_adb_command,
timeout_seconds: float = DEFAULT_ADB_TIMEOUT_SECONDS,
sleeper: Sleeper = time.sleep,
adb_executable: Optional[Path] = None,
):
if timeout_seconds <= 0:
raise ValueError("ADB 超时时间必须大于 0")
self._run_command = command_runner
self._timeout_seconds = timeout_seconds
self._sleep = sleeper
self._uses_bundled_adb = adb_executable is None
self._adb_executable = str(adb_executable or bundled_adb_path())
def search(
self, is_cancelled: Optional[Callable[[], bool]] = None
) -> List[AndroidDevice]:
"""返回当前 ADB 设备；取消后停止补充设备属性。"""
result = self._run(["adb", "devices", "-l"], "搜索设备")
result = self._run(
[self._adb_executable, "devices", "-l"], "搜索设备"
)
devices = self.parse_devices(result.stdout or "")
enriched = []
@@ -158,7 +170,15 @@ class AndroidDeviceService:
raise AndroidDeviceSearchError("应用包名格式不正确")
result = self._run(
["adb", "-s", serial, "shell", "pm", "path", package_name],
[
self._adb_executable,
"-s",
serial,
"shell",
"pm",
"path",
package_name,
],
"检测 PDD 应用",
)
return any(
@@ -193,7 +213,7 @@ class AndroidDeviceService:
self._check_cancelled(is_cancelled)
self._report(on_progress, "正在读取勾选设备的 Wi-Fi 地址…")
route = self._run(
["adb", "-s", usb_serial, "shell", "ip", "route"],
[self._adb_executable, "-s", usb_serial, "shell", "ip", "route"],
"读取 Wi-Fi 地址",
)
ip_address = self.parse_wifi_ipv4(route.stdout or "")
@@ -202,7 +222,7 @@ class AndroidDeviceService:
self._check_cancelled(is_cancelled)
self._report(on_progress, "正在开启 ADB 5555 端口…")
self._run(
["adb", "-s", usb_serial, "tcpip", "5555"],
[self._adb_executable, "-s", usb_serial, "tcpip", "5555"],
"开启 Wi-Fi 调试",
)
@@ -312,7 +332,14 @@ class AndroidDeviceService:
try:
result = self._run(
["adb", "-s", serial, "shell", "getprop", property_name],
[
self._adb_executable,
"-s",
serial,
"shell",
"getprop",
property_name,
],
"读取设备信息",
)
except AndroidDeviceSearchError:
@@ -343,12 +370,14 @@ class AndroidDeviceService:
self._sleep(DEFAULT_DEVICE_READY_INTERVAL_SECONDS)
def _list_devices(self) -> List[AndroidDevice]:
result = self._run(["adb", "devices", "-l"], "读取设备列表")
result = self._run(
[self._adb_executable, "devices", "-l"], "读取设备列表"
)
return self.parse_devices(result.stdout or "")
def _connect_wifi(self, wifi_serial: str) -> None:
result = self._run(
["adb", "connect", wifi_serial],
[self._adb_executable, "connect", wifi_serial],
"连接 Wi-Fi 设备",
)
detail = " ".join(
@@ -401,11 +430,16 @@ class AndroidDeviceService:
callback(message)
def _run(self, command: Sequence[str], action: str):
if self._uses_bundled_adb:
try:
validate_bundled_adb()
except BundledAdbError as exc:
raise AndroidDeviceSearchError(str(exc)) from exc
try:
result = self._run_command(command, self._timeout_seconds)
except FileNotFoundError as exc:
raise AndroidDeviceSearchError(
"未找到 adb，请安装 Android platform-tools 并把 adb 加入 PATH 后重试"
"内置 adb.exe 无法启动，请重新安装完整的软件包"
) from exc
except subprocess.TimeoutExpired as exc:
raise AndroidDeviceSearchError(
@@ -413,7 +447,7 @@ class AndroidDeviceService:
) from exc
except OSError as exc:
raise AndroidDeviceSearchError(
f"无法启动 adb：{str(exc) or '系统拒绝执行'}"
f"无法启动内置 adb：{str(exc) or '系统拒绝执行'}"
) from exc
if result.returncode != 0:
client/src/db.py
+8
View File
@@ -34,6 +34,14 @@ def data_dir() -> Path:
return directory
def app_dir() -> Path:
"""返回只读资源根目录；源码和打包运行使用同一相对结构。"""
if getattr(sys, "frozen", False):
return Path(sys._MEIPASS).resolve()
return Path(__file__).resolve().parents[1]
def default_database_path() -> Path:
"""返回默认数据库文件路径。"""
client/src/pdd_device_service.py
+5
View File
@@ -12,6 +12,7 @@ from contextlib import AbstractContextManager, contextmanager
from contextvars import ContextVar
from typing import Any, Callable, Iterator, Optional
from .adb_runtime import BundledAdbError, configure_bundled_adb_environment
from .performance_timing import current_performance_trace
@@ -80,6 +81,10 @@ class PddDeviceService:
@staticmethod
def _default_connector(serial: str) -> Any:
try:
configure_bundled_adb_environment()
except BundledAdbError as exc:
raise PddDeviceError("DEVICE_ADB_MISSING", str(exc)) from exc
try:
import uiautomator2 as u2
except ImportError as exc:
client/src/settings_ui_event.py
+14 -2
View File
@@ -15,6 +15,7 @@ from PyQt5.QtCore import (
pyqtSlot,
)
from .adb_runtime import BundledAdbError, validate_bundled_adb
from .android_device_service import (
AndroidDeviceConversionCancelled,
AndroidDeviceSearchError,
@@ -359,6 +360,12 @@ class SettingsPageEventBinder(QObject):
self._live_purchase_adapter_ready = bool(
live_purchase_adapter_ready
)
self._adb_unavailable_message = ""
if android_device_service is None:
try:
validate_bundled_adb()
except BundledAdbError as exc:
self._adb_unavailable_message = str(exc)
self._android_device_service = (
android_device_service or AndroidDeviceService()
)
@@ -416,8 +423,12 @@ class SettingsPageEventBinder(QObject):
self._load_current_client()
self._load_selected_android_device()
if self._adb_unavailable_message:
self._page.deviceStatusLabel.setText(
self._adb_unavailable_message
)
self._sync_button_state()
if self._saved_android_serial:
if self._saved_android_serial and not self._adb_unavailable_message:
QTimer.singleShot(0, self._request_restore_saved_android_device)
@pyqtSlot()
@@ -910,7 +921,8 @@ class SettingsPageEventBinder(QObject):
and not self._android_setting_busy
)
device_commands_enabled = (
not self._busy
not self._adb_unavailable_message
and not self._busy
and not self._search_busy
and not self._wifi_conversion_busy
and not self._android_setting_busy
client/src/ui_main.py
+24 -1
View File
@@ -17,16 +17,18 @@
import sys
from PyQt5.QtCore import Qt
from PyQt5.QtCore import Qt, QTimer
from PyQt5.QtWidgets import QApplication
from qfluentwidgets import (
FluentIcon as FIF,
FluentWindow,
MessageBox,
NavigationItemPosition,
Theme,
setTheme,
)
from .adb_runtime import BundledAdbError, configure_bundled_adb_environment
from .pdd_ui import PDDTaskPage
from .pdd_ui_event import PDDTaskPageEvent
from .pdd_u2_purchase_adapter import (
@@ -147,8 +149,19 @@ def ui_main():
app = QApplication(sys.argv)
setTheme(Theme.AUTO)
adb_error = ""
try:
configure_bundled_adb_environment()
except BundledAdbError as exc:
adb_error = str(exc)
window = MainWindow()
window.showMaximized()
if adb_error:
QTimer.singleShot(
0,
lambda: _show_adb_missing_dialog(window, adb_error),
)
try:
mark_current_version_healthy()
except OSError:
@@ -157,5 +170,15 @@ def ui_main():
return app.exec_()
def _show_adb_missing_dialog(parent: MainWindow, message: str) -> None:
"""在主窗口中央提示内置 ADB 缺失，用户可用大按钮关闭。"""
dialog = MessageBox("Android 调试组件缺失", message, parent)
dialog.yesButton.setText("知道了")
dialog.yesButton.setMinimumSize(120, 40)
dialog.cancelButton.hide()
dialog.exec()
if __name__ == "__main__":
sys.exit(ui_main())