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+)\]$") 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"") 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"")]) 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") # The real MEmu device uses MemuIME. Android's generic ``input text`` # drops non-ASCII search terms, so replace the focused field through the # emulator's UTF-8/Base64 clipboard service and paste key. The repeated # keycodes are a single argv invocation (not a remote shell script). end = self.run("shell", "input", "keyevent", "123") clear = self.run("shell", "input", "keyevent", *("67" for _ in range(256))) encoded = base64.b64encode(value.encode("utf-8")).decode("ascii") clipboard = self.run( "shell", "am", "startservice", "-a", "memu_clip_board_base64", "--es", "clip", encoded, ) paste = self.run("shell", "input", "keyevent", "279") if any(result.exit_code != 0 for result in (end, clear, clipboard, paste)): raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "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: raise ContractError(ErrorCode.PACKAGE_MISSING, "start_package", "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")