72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""共享的 Android 工具函数，可被框架内外复用。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import time
|
|
from typing import Callable, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_current_package(
|
|
adb_shell_func: Callable[[str], str],
|
|
max_attempts: int = 3,
|
|
retry_delay: float = 0.5,
|
|
) -> Optional[str]:
|
|
"""Read the foreground package via ``dumpsys``, retrying transient failures.
|
|
|
|
Args:
|
|
adb_shell_func: A callable that accepts a shell command string and
|
|
returns the stdout text. Example::
|
|
|
|
lambda cmd: adb_client.shell(cmd)
|
|
max_attempts: Maximum retry attempts for transient failures.
|
|
retry_delay: Seconds to wait between retries.
|
|
|
|
Returns:
|
|
The foreground package name, or ``None`` if it could not be determined.
|
|
"""
|
|
for attempt in range(max_attempts):
|
|
try:
|
|
focused_app = adb_shell_func("dumpsys window | grep mFocusedApp")
|
|
if focused_app:
|
|
matches = re.findall(
|
|
r"mFocusedApp=ActivityRecord\{[a-f0-9]+\s+\S+\s+([^/]+)/",
|
|
focused_app,
|
|
)
|
|
if matches:
|
|
return matches[-1]
|
|
|
|
current_focus = adb_shell_func("dumpsys window | grep mCurrentFocus")
|
|
if current_focus:
|
|
matches = re.findall(
|
|
r"Window\{[a-f0-9]+\s+\S+\s+([^/]+)/",
|
|
current_focus,
|
|
)
|
|
if matches:
|
|
return matches[-1]
|
|
|
|
resumed_activity = adb_shell_func("dumpsys activity activities | grep mResumedActivity")
|
|
if resumed_activity:
|
|
matches = re.findall(
|
|
r"\{[a-f0-9]+\s+\S+\s+([^/]+)/",
|
|
resumed_activity,
|
|
)
|
|
if matches:
|
|
return matches[-1]
|
|
|
|
if attempt < max_attempts - 1:
|
|
time.sleep(retry_delay)
|
|
except Exception as exc:
|
|
if attempt < max_attempts - 1:
|
|
time.sleep(retry_delay)
|
|
continue
|
|
logger.error("Failed to get current package: %s", exc)
|
|
return None
|
|
|
|
logger.warning("Failed to get package name after %d attempts", max_attempts)
|
|
return None
|