471 lines
16 KiB
Python
471 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
import shutil
|
|
import tempfile
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from urllib.parse import quote
|
|
|
|
|
|
def adb_available() -> bool:
|
|
return shutil.which("adb") is not None
|
|
|
|
|
|
async def _run_cmd(
|
|
command: list[str],
|
|
*,
|
|
timeout: float = 20.0,
|
|
capture_binary: bool = False,
|
|
) -> tuple[int, str | bytes, str]:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*command,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
try:
|
|
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
|
|
except asyncio.TimeoutError:
|
|
proc.kill()
|
|
await proc.wait()
|
|
return 124, b"" if capture_binary else "", "Command timed out"
|
|
|
|
if capture_binary:
|
|
return proc.returncode or 0, stdout or b"", (stderr or b"").decode("utf-8", "replace").strip()
|
|
return (
|
|
proc.returncode or 0,
|
|
(stdout or b"").decode("utf-8", "replace").strip(),
|
|
(stderr or b"").decode("utf-8", "replace").strip(),
|
|
)
|
|
|
|
|
|
def _adb_prefix(serial: Optional[str]) -> list[str]:
|
|
command = ["adb"]
|
|
if serial:
|
|
command.extend(["-s", serial])
|
|
return command
|
|
|
|
|
|
async def _adb_text(serial: Optional[str], *args: str, timeout: float = 20.0) -> tuple[int, str, str]:
|
|
code, out, err = await _run_cmd(_adb_prefix(serial) + list(args), timeout=timeout)
|
|
assert isinstance(out, str)
|
|
return code, out, err
|
|
|
|
|
|
async def _adb_binary(serial: Optional[str], *args: str, timeout: float = 20.0) -> tuple[int, bytes, str]:
|
|
code, out, err = await _run_cmd(
|
|
_adb_prefix(serial) + list(args),
|
|
timeout=timeout,
|
|
capture_binary=True,
|
|
)
|
|
assert isinstance(out, bytes)
|
|
return code, out, err
|
|
|
|
|
|
def _parse_device_line(line: str) -> Optional[dict]:
|
|
line = line.strip()
|
|
if not line or line.startswith("List of devices attached"):
|
|
return None
|
|
parts = line.split()
|
|
if len(parts) < 2:
|
|
return None
|
|
|
|
serial = parts[0]
|
|
state = parts[1]
|
|
extras: dict[str, str] = {}
|
|
for token in parts[2:]:
|
|
if ":" not in token:
|
|
continue
|
|
key, value = token.split(":", 1)
|
|
extras[key] = value
|
|
|
|
return {
|
|
"serial": serial,
|
|
"state": state,
|
|
"usb": extras.get("usb", ""),
|
|
"product": extras.get("product", ""),
|
|
"model": extras.get("model", ""),
|
|
"device": extras.get("device", ""),
|
|
"transport_id": extras.get("transport_id", ""),
|
|
"raw": line,
|
|
}
|
|
|
|
|
|
async def list_android_devices() -> list[dict]:
|
|
if not adb_available():
|
|
return []
|
|
code, out, _ = await _adb_text(None, "devices", "-l")
|
|
if code != 0:
|
|
return []
|
|
devices: list[dict] = []
|
|
for raw in out.splitlines():
|
|
parsed = _parse_device_line(raw)
|
|
if parsed:
|
|
devices.append(parsed)
|
|
return devices
|
|
|
|
|
|
async def _shell(serial: str, command: str, timeout: float = 20.0) -> str:
|
|
code, out, err = await _adb_text(serial, "shell", command, timeout=timeout)
|
|
if code != 0:
|
|
raise RuntimeError(err or out or "ADB shell command failed")
|
|
return out or err
|
|
|
|
|
|
async def _get_display_size(serial: str) -> tuple[int, int]:
|
|
output = await _shell(serial, "wm size | tail -n 1")
|
|
match = re.search(r"(\d+)x(\d+)", output)
|
|
if not match:
|
|
return 1080, 2400
|
|
return int(match.group(1)), int(match.group(2))
|
|
|
|
|
|
async def android_device_overview(serial: str) -> dict:
|
|
if not serial:
|
|
raise ValueError("Missing Android device serial")
|
|
|
|
props = {
|
|
"model": "getprop ro.product.model",
|
|
"brand": "getprop ro.product.brand",
|
|
"android_version": "getprop ro.build.version.release",
|
|
"sdk": "getprop ro.build.version.sdk",
|
|
}
|
|
|
|
async def grab(name: str, cmd: str) -> tuple[str, str]:
|
|
try:
|
|
return name, (await _shell(serial, cmd, timeout=10.0)).strip()
|
|
except RuntimeError as exc:
|
|
return name, str(exc)
|
|
|
|
results = dict(await asyncio.gather(*(grab(key, cmd) for key, cmd in props.items())))
|
|
|
|
battery = await _shell(
|
|
serial,
|
|
"dumpsys battery | grep -E 'level|status|powered' | sed 's/^[[:space:]]*//'",
|
|
timeout=10.0,
|
|
)
|
|
focus = await _shell(
|
|
serial,
|
|
"(dumpsys window windows | grep -E 'mCurrentFocus' | tail -n 1) || "
|
|
"(dumpsys activity activities | grep -E 'mResumedActivity' | tail -n 1)",
|
|
timeout=10.0,
|
|
)
|
|
resolution = await _shell(serial, "wm size | tail -n 1", timeout=10.0)
|
|
rotation = await _shell(serial, "settings get system user_rotation", timeout=10.0)
|
|
screen_on = await _shell(serial, "dumpsys power | grep -E 'Display Power|mHoldingDisplaySuspendBlocker'", timeout=10.0)
|
|
|
|
return {
|
|
"serial": serial,
|
|
"model": results.get("model", ""),
|
|
"brand": results.get("brand", ""),
|
|
"android_version": results.get("android_version", ""),
|
|
"sdk": results.get("sdk", ""),
|
|
"battery": battery,
|
|
"focus": focus,
|
|
"resolution": resolution,
|
|
"rotation": rotation,
|
|
"screen_state": screen_on,
|
|
}
|
|
|
|
|
|
def _bounds_to_rect(bounds: str) -> Optional[tuple[int, int, int, int]]:
|
|
match = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds.strip())
|
|
if not match:
|
|
return None
|
|
left, top, right, bottom = (int(match.group(i)) for i in range(1, 5))
|
|
if right <= left or bottom <= top:
|
|
return None
|
|
return left, top, right, bottom
|
|
|
|
|
|
def _node_label(node: ET.Element) -> str:
|
|
text = (node.attrib.get("text") or "").strip()
|
|
desc = (node.attrib.get("content-desc") or "").strip()
|
|
resource_id = (node.attrib.get("resource-id") or "").strip()
|
|
class_name = (node.attrib.get("class") or "").strip()
|
|
if text:
|
|
return text
|
|
if desc:
|
|
return desc
|
|
if resource_id:
|
|
return resource_id.split("/")[-1]
|
|
return class_name.split(".")[-1] if class_name else "node"
|
|
|
|
|
|
async def android_packages(serial: str, query: str = "", limit: int = 60) -> list[dict]:
|
|
if not serial:
|
|
raise ValueError("Missing Android device serial")
|
|
|
|
output = await _shell(serial, "pm list packages", timeout=25.0)
|
|
needle = query.strip().lower()
|
|
results: list[dict] = []
|
|
for raw in output.splitlines():
|
|
package = raw.strip()
|
|
if package.startswith("package:"):
|
|
package = package[len("package:") :]
|
|
if not package:
|
|
continue
|
|
if needle and needle not in package.lower():
|
|
continue
|
|
results.append(
|
|
{
|
|
"package": package,
|
|
"label": package.split(".")[-1],
|
|
}
|
|
)
|
|
if len(results) >= max(1, min(limit, 200)):
|
|
break
|
|
return results
|
|
|
|
|
|
async def android_ui_nodes(serial: str, limit: int = 80) -> list[dict]:
|
|
if not serial:
|
|
raise ValueError("Missing Android device serial")
|
|
|
|
xml_text = ""
|
|
for dump_path in ("/data/local/tmp/live-ui.xml", "/sdcard/live-ui.xml"):
|
|
try:
|
|
xml_text = await _shell(
|
|
serial,
|
|
f"uiautomator dump {dump_path} >/dev/null 2>&1 && cat {dump_path}",
|
|
timeout=25.0,
|
|
)
|
|
except RuntimeError:
|
|
xml_text = ""
|
|
if xml_text.strip().startswith("<?xml"):
|
|
break
|
|
xml_text = xml_text.strip()
|
|
if not xml_text.startswith("<?xml"):
|
|
raise RuntimeError("Failed to dump Android UI hierarchy")
|
|
|
|
try:
|
|
root = ET.fromstring(xml_text)
|
|
except ET.ParseError as exc:
|
|
raise RuntimeError(f"Unable to parse Android UI dump: {exc}") from exc
|
|
|
|
nodes: list[dict] = []
|
|
for node in root.iter("node"):
|
|
bounds = (node.attrib.get("bounds") or "").strip()
|
|
rect = _bounds_to_rect(bounds)
|
|
if not rect:
|
|
continue
|
|
|
|
text = (node.attrib.get("text") or "").strip()
|
|
desc = (node.attrib.get("content-desc") or "").strip()
|
|
resource_id = (node.attrib.get("resource-id") or "").strip()
|
|
clickable = node.attrib.get("clickable") == "true"
|
|
focusable = node.attrib.get("focusable") == "true"
|
|
scrollable = node.attrib.get("scrollable") == "true"
|
|
enabled = node.attrib.get("enabled") != "false"
|
|
if not (text or desc or resource_id or clickable or focusable or scrollable):
|
|
continue
|
|
|
|
left, top, right, bottom = rect
|
|
center_x = left + (right - left) // 2
|
|
center_y = top + (bottom - top) // 2
|
|
nodes.append(
|
|
{
|
|
"label": _node_label(node),
|
|
"text": text,
|
|
"content_desc": desc,
|
|
"resource_id": resource_id,
|
|
"class_name": (node.attrib.get("class") or "").strip(),
|
|
"package": (node.attrib.get("package") or "").strip(),
|
|
"bounds": bounds,
|
|
"center_x": center_x,
|
|
"center_y": center_y,
|
|
"clickable": clickable,
|
|
"focusable": focusable,
|
|
"scrollable": scrollable,
|
|
"enabled": enabled,
|
|
}
|
|
)
|
|
|
|
def sort_key(item: dict) -> tuple[int, int, int]:
|
|
return (
|
|
0 if item["clickable"] else 1,
|
|
0 if item["text"] or item["content_desc"] else 1,
|
|
len(item["label"]),
|
|
)
|
|
|
|
nodes.sort(key=sort_key)
|
|
return nodes[: max(1, min(limit, 200))]
|
|
|
|
|
|
def _is_png(data: bytes) -> bool:
|
|
return bool(data and len(data) > 8 and data.startswith(b"\x89PNG\r\n\x1a\n"))
|
|
|
|
|
|
async def android_logcat_recent(serial: str, lines: int = 200) -> str:
|
|
"""Dump recent logcat lines (logcat -d -t N), similar to device-console tooling."""
|
|
if not serial:
|
|
raise ValueError("Missing Android device serial")
|
|
n = max(1, min(int(lines), 3000))
|
|
return await _shell(serial, f"logcat -d -t {n}", timeout=60.0)
|
|
|
|
|
|
async def android_screenshot(serial: str) -> bytes:
|
|
if not serial:
|
|
raise ValueError("Missing Android device serial")
|
|
code, data, err = await _adb_binary(serial, "exec-out", "screencap", "-p", timeout=20.0)
|
|
if code == 0 and data:
|
|
normalized = data.replace(b"\r\n", b"\n")
|
|
if _is_png(data) or normalized.startswith(b"\x89PNG"):
|
|
return normalized
|
|
|
|
last_err = err or "exec-out screencap failed"
|
|
for remote in ("/data/local/tmp/live-cap.png", "/sdcard/live-cap.png"):
|
|
tmp = Path(tempfile.mkstemp(suffix=".png")[1])
|
|
try:
|
|
sh_code, _, sh_err = await _adb_text(serial, "shell", f"screencap -p {remote}", timeout=25.0)
|
|
if sh_code != 0:
|
|
last_err = sh_err or f"screencap to {remote} failed"
|
|
continue
|
|
pull_code, _, pull_err = await _run_cmd(
|
|
_adb_prefix(serial) + ["pull", remote, str(tmp)],
|
|
timeout=35.0,
|
|
)
|
|
if pull_code != 0:
|
|
last_err = pull_err or "adb pull failed"
|
|
continue
|
|
pulled = tmp.read_bytes()
|
|
if _is_png(pulled) or pulled.startswith(b"\x89PNG"):
|
|
return pulled.replace(b"\r\n", b"\n")
|
|
last_err = "Pulled file is not a valid PNG"
|
|
finally:
|
|
tmp.unlink(missing_ok=True)
|
|
|
|
raise RuntimeError(last_err)
|
|
|
|
|
|
def _normalize_text(text: str) -> str:
|
|
normalized = text.replace(" ", "%s")
|
|
normalized = normalized.replace("&", r"\&")
|
|
normalized = normalized.replace("(", r"\(")
|
|
normalized = normalized.replace(")", r"\)")
|
|
return normalized
|
|
|
|
|
|
async def android_action(action: str, serial: str, payload: Optional[dict] = None) -> dict:
|
|
if not serial:
|
|
raise ValueError("Missing Android device serial")
|
|
if not adb_available():
|
|
raise RuntimeError("adb is not installed")
|
|
payload = payload or {}
|
|
|
|
if action == "key":
|
|
keycode = str(payload.get("keycode", "")).strip()
|
|
if not keycode:
|
|
raise ValueError("Missing keycode")
|
|
await _shell(serial, f"input keyevent {keycode}")
|
|
return {"message": f"Sent keycode {keycode}"}
|
|
|
|
if action == "text":
|
|
text = str(payload.get("text", ""))
|
|
if not text:
|
|
raise ValueError("Missing text")
|
|
await _shell(serial, f"input text '{_normalize_text(text)}'")
|
|
return {"message": "Text sent to device"}
|
|
|
|
if action == "open_url":
|
|
url = str(payload.get("url", "")).strip()
|
|
if not url:
|
|
raise ValueError("Missing url")
|
|
await _shell(
|
|
serial,
|
|
f"am start -a android.intent.action.VIEW -d '{quote(url, safe=':/?&=%#.-_')}'",
|
|
)
|
|
return {"message": f"Opened URL: {url}"}
|
|
|
|
if action == "launch_package":
|
|
package = str(payload.get("package", "")).strip()
|
|
if not package:
|
|
raise ValueError("Missing package")
|
|
await _shell(serial, f"monkey -p '{package}' -c android.intent.category.LAUNCHER 1")
|
|
return {"message": f"Launched package: {package}"}
|
|
|
|
if action == "shell":
|
|
command = str(payload.get("command", "")).strip()
|
|
if not command:
|
|
raise ValueError("Missing shell command")
|
|
output = await _shell(serial, command, timeout=30.0)
|
|
return {"message": "Shell command finished", "output": output}
|
|
|
|
if action == "tap":
|
|
x = int(payload.get("x", 0))
|
|
y = int(payload.get("y", 0))
|
|
await _shell(serial, f"input tap {x} {y}")
|
|
return {"message": f"Tapped at {x}, {y}"}
|
|
|
|
if action == "double_tap":
|
|
x = int(payload.get("x", 0))
|
|
y = int(payload.get("y", 0))
|
|
await _shell(serial, f"input tap {x} {y} && sleep 0.08 && input tap {x} {y}")
|
|
return {"message": f"Double tapped at {x}, {y}"}
|
|
|
|
if action == "long_press":
|
|
x = int(payload.get("x", 0))
|
|
y = int(payload.get("y", 0))
|
|
duration = int(payload.get("duration", 700))
|
|
await _shell(serial, f"input swipe {x} {y} {x} {y} {duration}")
|
|
return {"message": f"Long pressed at {x}, {y} for {duration}ms"}
|
|
|
|
if action == "swipe":
|
|
direction = str(payload.get("direction", "")).strip().lower()
|
|
width, height = await _get_display_size(serial)
|
|
mid_x = max(width // 2, 1)
|
|
mid_y = max(height // 2, 1)
|
|
mapping = {
|
|
"up": (mid_x, int(height * 0.75), mid_x, int(height * 0.28)),
|
|
"down": (mid_x, int(height * 0.28), mid_x, int(height * 0.75)),
|
|
"left": (int(width * 0.75), mid_y, int(width * 0.25), mid_y),
|
|
"right": (int(width * 0.25), mid_y, int(width * 0.75), mid_y),
|
|
}
|
|
if direction not in mapping:
|
|
raise ValueError("Unsupported swipe direction")
|
|
start_x, start_y, end_x, end_y = mapping[direction]
|
|
await _shell(serial, f"input swipe {start_x} {start_y} {end_x} {end_y} 240")
|
|
return {"message": f"Swiped {direction}"}
|
|
|
|
if action == "wake":
|
|
await _shell(serial, "input keyevent KEYCODE_WAKEUP")
|
|
return {"message": "Wake command sent"}
|
|
|
|
if action == "rotate":
|
|
target = str(payload.get("rotation", "0")).strip()
|
|
if target not in {"0", "1", "2", "3"}:
|
|
raise ValueError("Rotation must be one of 0,1,2,3")
|
|
await _shell(
|
|
serial,
|
|
"settings put system accelerometer_rotation 0 && "
|
|
f"settings put system user_rotation {target}",
|
|
)
|
|
return {"message": f"Rotation set to {target}"}
|
|
|
|
if action == "force_stop":
|
|
package = str(payload.get("package", "")).strip()
|
|
if not package:
|
|
raise ValueError("Missing package")
|
|
await _shell(serial, f"am force-stop '{package}'")
|
|
return {"message": f"Force-stopped: {package}"}
|
|
|
|
if action == "clear_app":
|
|
package = str(payload.get("package", "")).strip()
|
|
if not package:
|
|
raise ValueError("Missing package")
|
|
output = await _shell(serial, f"pm clear '{package}'", timeout=25.0)
|
|
return {"message": f"Cleared app data: {package}", "output": output}
|
|
|
|
if action == "install_apk":
|
|
apk_path = str(payload.get("path", "")).strip()
|
|
if not apk_path or any(ch in apk_path for ch in ";$`&|()<>\n\r"):
|
|
raise ValueError("Invalid apk path on device (use adb push to /sdcard/ first)")
|
|
apk_path = apk_path.replace("'", "'\\''")
|
|
output = await _shell(serial, f"pm install -r -t '{apk_path}'", timeout=180.0)
|
|
return {"message": "APK install finished", "output": output}
|
|
|
|
raise ValueError(f"Unsupported Android action: {action}")
|