114 lines
4.2 KiB
Python
114 lines
4.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""通用 ADB 命令封装，不依赖特定框架层级。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import subprocess
|
|
import time
|
|
from typing import List, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ADBClient:
|
|
"""通过 subprocess 调用 adb 的通用封装。
|
|
|
|
与具体框架解耦，仅依赖系统 PATH 中的 adb。
|
|
"""
|
|
|
|
def __init__(self, serial: Optional[str] = None, adb_path: str = "adb"):
|
|
"""
|
|
Args:
|
|
serial: 目标设备序列号。为 ``None`` 时使用 adb 默认设备。
|
|
adb_path: adb 可执行文件路径。
|
|
"""
|
|
self.serial = serial
|
|
self.adb_path = adb_path
|
|
self.logger = logging.getLogger(self.__class__.__name__)
|
|
|
|
def _build_cmd(self, args: List[str]) -> List[str]:
|
|
cmd = [self.adb_path]
|
|
if self.serial:
|
|
cmd.extend(["-s", self.serial])
|
|
cmd.extend(args)
|
|
return cmd
|
|
|
|
def run(
|
|
self,
|
|
args: List[str],
|
|
*,
|
|
check: bool = False,
|
|
timeout: int = 30,
|
|
retries: int = 1,
|
|
retry_delay: float = 1.0,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
"""执行 adb 命令并返回 :class:`subprocess.CompletedProcess`。
|
|
|
|
Args:
|
|
args: adb 子命令及参数列表，如 ``["shell", "ls", "/sdcard"]``
|
|
check: 若为 ``True``，返回码非零时抛出 :exc:`RuntimeError`。
|
|
timeout: 单条命令超时秒数。
|
|
retries: 最大重试次数（仅对"空输出 + 非零退出码"的瞬态错误重试）。
|
|
retry_delay: 重试间隔秒数。
|
|
|
|
Returns:
|
|
命令执行结果。
|
|
|
|
Raises:
|
|
RuntimeError: adb 未找到，或 ``check=True`` 且命令最终失败。
|
|
"""
|
|
cmd = self._build_cmd(args)
|
|
last_result: Optional[subprocess.CompletedProcess[str]] = None
|
|
|
|
for attempt in range(1, retries + 1):
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
check=False,
|
|
timeout=max(timeout, 1),
|
|
)
|
|
except FileNotFoundError as exc:
|
|
raise RuntimeError("adb not found. Please ensure adb is installed and in PATH.") from exc
|
|
|
|
last_result = result
|
|
if result.returncode == 0:
|
|
return result
|
|
|
|
# If there is meaningful output, no need to retry — the command
|
|
# executed but reported an error (e.g. package not found).
|
|
stdout = (result.stdout or "").strip()
|
|
stderr = (result.stderr or "").strip()
|
|
if stdout or stderr:
|
|
break
|
|
|
|
# Empty output with non-zero exit code is likely a transient ADB
|
|
# connection issue; retry after a short delay.
|
|
if attempt < retries:
|
|
self.logger.warning(
|
|
"ADB command returned %d with no output, retrying (%d/%d): %s",
|
|
result.returncode, attempt, retries, " ".join(cmd),
|
|
)
|
|
time.sleep(retry_delay)
|
|
|
|
assert last_result is not None
|
|
if check and last_result.returncode != 0:
|
|
error = (last_result.stderr or last_result.stdout or "adb command failed").strip()
|
|
raise RuntimeError(f"ADB command failed ({' '.join(cmd)}): {error}")
|
|
return last_result
|
|
|
|
def shell(self, command: str | List[str], *, check: bool = False, timeout: int = 30) -> str:
|
|
"""执行 ``adb shell <command>`` 并返回 stdout 文本。"""
|
|
if isinstance(command, list):
|
|
# Pass arguments directly; adb will forward them to the device shell.
|
|
return self.run(["shell"] + command, check=check, timeout=timeout).stdout.strip()
|
|
return self.run(["shell", command], check=check, timeout=timeout).stdout.strip()
|
|
|
|
def pull(self, remote: str, local: str, *, check: bool = True, timeout: int = 30) -> str:
|
|
"""执行 ``adb pull`` 并返回 stdout 文本。"""
|
|
return self.run(["pull", "-a", remote, local], check=check, timeout=timeout).stdout.strip()
|