97 lines
4.3 KiB
Python
97 lines
4.3 KiB
Python
"""Launch Android activities through Activity Manager, without requiring Monkey."""
|
|
|
|
import asyncio
|
|
import re
|
|
|
|
from adbutils import AdbDevice
|
|
|
|
from artemis.drivers.errors import AppLaunchError
|
|
|
|
_PACKAGE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*")
|
|
_COMPONENT = re.compile(r"([\w.]+)/([\w.$]+)", re.ASCII)
|
|
_ERROR = re.compile(
|
|
r"^\s*(?:error\b|exception\b|securityexception\b|permission denial\b|"
|
|
r"[\w.]+Exception\b)",
|
|
re.IGNORECASE | re.MULTILINE,
|
|
)
|
|
_TRANSIENT = re.compile(r"timed? out|timeout|deadobjectexception|can't find service", re.IGNORECASE)
|
|
_UNSUPPORTED = re.compile(
|
|
r"unknown (?:command|option)|unrecognized (?:command|option)", re.IGNORECASE
|
|
)
|
|
|
|
|
|
def _failure(package: str, stage: str, code: int, output: str, **kwargs: bool) -> AppLaunchError:
|
|
return AppLaunchError(
|
|
f"应用 {package} {stage}失败（退出码 {code}）：{output.strip()[:1500] or '命令没有返回结果'}",
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
async def _shell(
|
|
device: AdbDevice, command: list[str], package: str, stage: str, timeout: float
|
|
) -> tuple[int, str]:
|
|
try:
|
|
result = await asyncio.to_thread(device.shell2, command, timeout=timeout)
|
|
return result.returncode, result.output
|
|
except Exception as exc:
|
|
# A transport timeout can happen after Android has already received start.
|
|
# The caller must inspect the foreground before issuing another launch.
|
|
raise AppLaunchError(
|
|
f"应用 {package} {stage}时 ADB 请求失败：{str(exc)[:1500] or type(exc).__name__}",
|
|
retryable=True,
|
|
may_have_started=stage == "启动",
|
|
) from exc
|
|
|
|
|
|
async def launch_android_app(device: AdbDevice, package: str, timeout: float) -> bool:
|
|
"""Resolve the current user's launcher and request a normal launcher-style start.
|
|
|
|
True means Activity Manager accepted the command; the shared launch workflow
|
|
still verifies the foreground. Deterministic failures raise AppLaunchError so
|
|
they do not turn into repeated foreground timeouts.
|
|
"""
|
|
if not _PACKAGE.fullmatch(package):
|
|
raise AppLaunchError(f"无效的 Android 应用包名：{package!r}")
|
|
|
|
intent = ["-a", "android.intent.action.MAIN", "-c", "android.intent.category.LAUNCHER"]
|
|
resolve = ["resolve-activity", "--brief", "--user", "current", *intent, "-p", package]
|
|
component = None
|
|
# pm is the compatibility path for systems without cmd package. If neither
|
|
# exposes resolution, let Activity Manager resolve the package-scoped intent.
|
|
for prefix in (["cmd", "package"], ["pm"]):
|
|
code, output = await _shell(device, [*prefix, *resolve], package, "查询启动入口", timeout)
|
|
if code == 127 or _UNSUPPORTED.search(output):
|
|
continue
|
|
if code != 0 or _ERROR.search(output):
|
|
raise _failure(
|
|
package, "查询启动入口", code, output, retryable=bool(_TRANSIENT.search(output))
|
|
)
|
|
candidates = [
|
|
match.group(0)
|
|
for line in output.splitlines()
|
|
if (match := _COMPONENT.fullmatch(line.strip())) and match.group(1) == package
|
|
]
|
|
if len(candidates) != 1:
|
|
raise _failure(
|
|
package,
|
|
"查询启动入口",
|
|
code,
|
|
f"未找到唯一可启动入口；应用可能未安装、已停用或没有桌面入口。{output.strip()}",
|
|
)
|
|
component = candidates[0]
|
|
break
|
|
|
|
target = ["-n", component] if component else ["-p", package]
|
|
# Match a launcher tap: NEW_TASK | RESET_TASK_IF_NEEDED, preserving warm tasks.
|
|
command = ["am", "start", "-W", "--user", "current", *intent, "-f", "0x10200000", *target]
|
|
code, output = await _shell(device, command, package, "启动", timeout)
|
|
status = re.search(r"^\s*Status:\s*(\S+)", output, re.IGNORECASE | re.MULTILINE)
|
|
if code != 0 or _ERROR.search(output) or (status and status[1].lower() != "ok"):
|
|
transient = bool(_TRANSIENT.search(output))
|
|
raise _failure(
|
|
package, "启动", code, output, retryable=transient, may_have_started=transient
|
|
)
|
|
if not status and not re.search(r"Starting:\s*Intent|Warning:\s*Activity not started", output):
|
|
raise _failure(package, "启动", code, output, retryable=True, may_have_started=True)
|
|
return True
|