from __future__ import annotations
|
|
import base64
|
from dataclasses import dataclass
|
from pathlib import Path, PurePosixPath
|
import re
|
import shlex
|
import time
|
import xml.etree.ElementTree as ET
|
from typing import Callable
|
|
from .models import ContractError, ErrorCode
|
from .process import ProcessResult, ProcessSupervisor
|
|
|
@dataclass(frozen=True)
|
class DeviceSnapshot:
|
serial: str
|
state: str
|
package_installed: bool
|
foreground_package: str | None
|
cache_readable: bool
|
|
|
@dataclass(frozen=True)
|
class RemoteFile:
|
path: str
|
name: str
|
bytes: int
|
mtime_token: str
|
|
|
@dataclass(frozen=True)
|
class UiNode:
|
text: str
|
resource_id: str
|
class_name: str
|
bounds: tuple[int, int, int, int]
|
clickable: bool
|
|
|
BOUNDS_RE = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
|
SAFE_ANDROID_INPUT_TEXT_RE = re.compile(r"^[A-Za-z0-9._-]+$")
|
MEMU_IME_COMPONENT = "com.microvirt.memuime/.MemuIME"
|
WELCOME_ACTIVITY = "cn.com.hibor/.WelcomeActivity"
|
|
|
class AdbClient:
|
def __init__(self, executable: Path, supervisor: ProcessSupervisor, *, serial: str | None = None,
|
package_name: str = "cn.com.hibor",
|
cache_root: str = "/sdcard/Android/data/cn.com.hibor/files/myfile/",
|
timeout_provider: Callable[[str, int], int] | None = None):
|
self.executable = Path(executable)
|
self.supervisor = supervisor
|
self.serial = serial
|
self.package_name = package_name
|
self.cache_root = cache_root
|
self.timeout_provider = timeout_provider
|
if package_name != "cn.com.hibor" or cache_root != "/sdcard/Android/data/cn.com.hibor/files/myfile/":
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "adb", "package/cache boundary")
|
|
def _argv(self, *args: str) -> tuple[str, ...]:
|
prefix = (str(self.executable),)
|
if self.serial:
|
prefix += ("-s", self.serial)
|
return prefix + tuple(args)
|
|
def run(self, *args: str, timeout_ms: int = 15_000) -> ProcessResult:
|
if self.timeout_provider is not None:
|
timeout_ms = self.timeout_provider("adb:" + " ".join(args[:3]), timeout_ms)
|
result = self.supervisor.run(self._argv(*args), timeout_ms=timeout_ms)
|
if result.error_code:
|
raise ContractError(result.error_code, "adb", "process failure")
|
return result
|
|
def set_timeout_provider(self, provider: Callable[[str, int], int]) -> None:
|
self.timeout_provider = provider
|
|
def preflight(self) -> DeviceSnapshot:
|
timeout = 10_000
|
if self.timeout_provider is not None:
|
timeout = self.timeout_provider("adb:devices", timeout)
|
devices = self.supervisor.run((str(self.executable), "devices", "-l"), timeout_ms=timeout)
|
if devices.error_code or devices.exit_code != 0:
|
raise ContractError(ErrorCode.DEVICE_NOT_ONLINE, "adb devices", "failed")
|
rows = []
|
for line in devices.stdout_bytes.decode("utf-8", "replace").splitlines()[1:]:
|
parts = line.split()
|
if len(parts) >= 2 and not line.startswith("*"):
|
rows.append((parts[0], parts[1]))
|
online = [row for row in rows if row[1] == "device"]
|
if self.serial:
|
online = [row for row in online if row[0] == self.serial]
|
if len(online) != 1:
|
raise ContractError(ErrorCode.DEVICE_NOT_UNIQUE, "adb devices", f"online={len(online)}")
|
self.serial = online[0][0]
|
package = self.run("shell", "pm", "path", self.package_name)
|
installed = package.exit_code == 0 and package.stdout_bytes.startswith(b"package:")
|
if not installed:
|
raise ContractError(ErrorCode.PACKAGE_MISSING, "package", self.package_name)
|
cache = self.run("shell", "ls", "-ld", self.cache_root)
|
readable = cache.exit_code == 0
|
if not readable:
|
raise ContractError(ErrorCode.CACHE_UNREADABLE, "cache", self.cache_root)
|
focus = self.run("shell", "dumpsys", "window", "windows")
|
text = focus.stdout_bytes.decode("utf-8", "replace")
|
match = re.search(r"mCurrentFocus=.*?\s([A-Za-z0-9_.]+)/", text)
|
return DeviceSnapshot(self.serial, "ONLINE", True, match.group(1) if match else None, True)
|
|
def list_cache(self) -> tuple[RemoteFile, ...]:
|
result = self.run("shell", "ls", "-ln", self.cache_root)
|
if result.exit_code != 0:
|
raise ContractError(ErrorCode.CACHE_UNREADABLE, "cache", "ls failed")
|
files: list[RemoteFile] = []
|
for raw in result.stdout_bytes.decode("utf-8", "replace").splitlines():
|
parts = raw.split(maxsplit=7)
|
if len(parts) < 8 or not parts[0].startswith("-"):
|
continue
|
try:
|
size = int(parts[4])
|
except ValueError:
|
continue
|
name = parts[7]
|
path = str(PurePosixPath(self.cache_root) / name)
|
files.append(RemoteFile(path, name, size, " ".join(parts[5:7])))
|
return tuple(sorted(files, key=lambda item: item.path))
|
|
def remote_sha256(self, remote_path: str) -> str:
|
self._require_cache_path(remote_path)
|
result = self.run("shell", "sha256sum", remote_path, timeout_ms=60_000)
|
if result.exit_code != 0:
|
raise ContractError(ErrorCode.HASH_MISMATCH, "remote_sha256", "sha256sum failed")
|
token = result.stdout_bytes.decode("ascii", "replace").split(maxsplit=1)[0].lower()
|
if not re.fullmatch(r"[0-9a-f]{64}", token):
|
raise ContractError(ErrorCode.HASH_MISMATCH, "remote_sha256", "invalid output")
|
return token
|
|
def pull(self, remote_path: str, local_staging: Path, *, timeout_ms: int = 60_000) -> ProcessResult:
|
self._require_cache_path(remote_path)
|
if local_staging.exists() or local_staging.is_symlink():
|
raise ContractError(ErrorCode.FINAL_PATH_CONFLICT, "local_staging", "must be absent")
|
local_staging.parent.mkdir(parents=True, exist_ok=True)
|
result = self.run("pull", remote_path, str(local_staging), timeout_ms=timeout_ms)
|
if result.exit_code != 0 or not local_staging.is_file():
|
raise ContractError(ErrorCode.PULL_FAILED, "adb pull", "failed")
|
return result
|
|
def ui_nodes(self) -> tuple[UiNode, ...]:
|
data = b""
|
start = end = -1
|
for attempt in range(10):
|
result = self.run("exec-out", "uiautomator", "dump", "/dev/tty", timeout_ms=15_000)
|
data = result.stdout_bytes
|
start = data.find(b"<?xml")
|
end = data.rfind(b"</hierarchy>")
|
if result.exit_code == 0 and start >= 0 and end >= 0:
|
break
|
if attempt < 9:
|
time.sleep(0.5)
|
else:
|
detail = "dump failed" if result.exit_code != 0 else "XML absent"
|
raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "uiautomator", detail)
|
root = ET.fromstring(data[start:end + len(b"</hierarchy>")])
|
nodes: list[UiNode] = []
|
for node in root.iter("node"):
|
match = BOUNDS_RE.match(node.attrib.get("bounds", ""))
|
if not match:
|
continue
|
nodes.append(UiNode(
|
node.attrib.get("text", "").strip(), node.attrib.get("resource-id", ""),
|
node.attrib.get("class", ""), tuple(map(int, match.groups())),
|
node.attrib.get("clickable") == "true",
|
))
|
return tuple(nodes)
|
|
def screenshot(self, target: Path) -> None:
|
if target.exists() or target.is_symlink():
|
raise ContractError(ErrorCode.FINAL_PATH_CONFLICT, "screenshot", "target exists")
|
result = self.run("exec-out", "screencap", "-p", timeout_ms=15_000)
|
if result.exit_code != 0 or not result.stdout_bytes.startswith(b"\x89PNG"):
|
raise ContractError(ErrorCode.PROCESS_START_FAILED, "screenshot", "invalid PNG")
|
from .archive import create_exclusive_bytes
|
create_exclusive_bytes(target, result.stdout_bytes)
|
|
def tap(self, x: int, y: int) -> None:
|
result = self.run("shell", "input", "tap", str(x), str(y))
|
if result.exit_code != 0:
|
raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "tap", "failed")
|
|
def swipe(self, x1: int, y1: int, x2: int, y2: int, duration_ms: int = 300) -> None:
|
result = self.run("shell", "input", "swipe", str(x1), str(y1), str(x2), str(y2), str(duration_ms))
|
if result.exit_code != 0:
|
raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "swipe", "failed")
|
|
def back(self) -> None:
|
result = self.run("shell", "input", "keyevent", "4")
|
if result.exit_code != 0:
|
raise ContractError(ErrorCode.UI_CURSOR_RESTORE_FAILED, "back", "failed")
|
|
def input_text(self, value: str) -> None:
|
if any(ch in value for ch in "\n\r\x00"):
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "query", "control character")
|
use_android_input = SAFE_ANDROID_INPUT_TEXT_RE.fullmatch(value) is not None
|
if not use_android_input:
|
ime_list = self.run("shell", "ime", "list", "-s")
|
installed_imes = {
|
line.strip()
|
for line in ime_list.stdout_bytes.decode("utf-8", "replace").splitlines()
|
if line.strip()
|
}
|
if ime_list.exit_code != 0 or MEMU_IME_COMPONENT not in installed_imes:
|
raise ContractError(
|
ErrorCode.UI_ANCHOR_DRIFT, "input_text",
|
"safe input unavailable: MemuIME unsupported",
|
)
|
|
end = self.run("shell", "input", "keyevent", "123")
|
clear = self.run("shell", "input", "keyevent", *("67" for _ in range(256)))
|
if end.exit_code != 0 or clear.exit_code != 0:
|
raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "failed")
|
|
if use_android_input:
|
typed = self.run("shell", "input", "text", value)
|
if typed.exit_code != 0:
|
raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "generic input failed")
|
return
|
|
encoded = base64.b64encode(value.encode("utf-8")).decode("ascii")
|
clipboard = self.run(
|
"shell", "am", "startservice", "-a", "memu_clip_board_base64",
|
"--es", "clip", encoded,
|
)
|
if clipboard.exit_code != 0:
|
# Never issue paste after a failed clipboard service call: on some
|
# emulators that would paste unrelated host clipboard contents.
|
raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "Memu clipboard failed")
|
paste = self.run("shell", "input", "keyevent", "279")
|
if paste.exit_code != 0:
|
raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "paste failed")
|
|
def start_package(self) -> None:
|
result = self.run("shell", "monkey", "-p", self.package_name, "-c", "android.intent.category.LAUNCHER", "1")
|
if result.exit_code == 0:
|
return
|
diagnostic = (result.stdout_bytes + b"\n" + result.stderr_bytes).decode("utf-8", "replace").casefold()
|
monkey_missing = result.exit_code == 127 or (
|
"monkey" in diagnostic and any(token in diagnostic for token in ("not found", "inaccessible"))
|
)
|
if not monkey_missing:
|
raise ContractError(ErrorCode.PACKAGE_MISSING, "start_package", "monkey failed")
|
fallback = self.run("shell", "am", "start", "-W", "-n", WELCOME_ACTIVITY)
|
fallback_text = (fallback.stdout_bytes + b"\n" + fallback.stderr_bytes).decode(
|
"utf-8", "replace"
|
).casefold()
|
if (fallback.exit_code != 0 or "error:" in fallback_text
|
or "error type" in fallback_text or "exception" in fallback_text):
|
raise ContractError(ErrorCode.PACKAGE_MISSING, "start_package", "WelcomeActivity failed")
|
|
def _require_cache_path(self, path: str) -> None:
|
root = PurePosixPath(self.cache_root)
|
candidate = PurePosixPath(path)
|
if candidate == root or root not in candidate.parents or ".." in candidate.parts:
|
raise ContractError(ErrorCode.CACHE_UNREADABLE, "remote_path", "outside cache")
|
