from __future__ import annotations import logging import os import re import time import xml.etree.ElementTree as ET from dataclasses import dataclass from typing import Optional from adb_client import ADBClient from android_utils import get_current_package logger = logging.getLogger(__name__) _BOUNDS_RE = re.compile(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]") STORE_GOOGLE_PLAY = "google_play" STORE_AURORA = "aurora" @dataclass(frozen=True) class UiNode: text: str desc: str package: str enabled: bool clickable: bool bounds: tuple[int, int, int, int] @property def center(self) -> tuple[int, int]: left, top, right, bottom = self.bounds return ((left + right) // 2, (top + bottom) // 2) def labels(self) -> tuple[str, ...]: values = [] if self.text.strip(): values.append(self.text.strip()) if self.desc.strip(): values.append(self.desc.strip()) return tuple(values) class GooglePlayDownloader: GOOGLE_PLAY_PACKAGE = "com.android.vending" def __init__(self, serial: Optional[str] = None, store_type: str = STORE_GOOGLE_PLAY): self._adb = ADBClient(serial=serial) self._serial = serial self._store_type = store_type if store_type == STORE_AURORA: self._store_package = "com.aurora.store" self._store_name = "Aurora Store" self._build_url = self._build_aurora_url else: self._store_package = "com.android.vending" self._store_name = "Google Play" self._build_url = self._build_google_play_url OPEN_TIMEOUT_SEC = 30 INSTALL_TIMEOUT_SEC = 200 POLL_INTERVAL_SEC = 2 # 页面无有效内容的最大等待时间（Aurora 找不到包名会一直卡在加载界面） LOADING_TIMEOUT_SEC = 30 # Aurora Store 点击 Install 后页面无变化的超时时间 # 正常情况点击 Install 后数秒内应出现系统安装弹窗或下载进度， # 若 Install 按钮始终存在且无任何变化，说明 Aurora 未能触发下载 AURORA_INSTALL_STUCK_SEC = 8 INSTALL_BUTTONS = ("Install", "Update") SUCCESS_BUTTONS = ("Open", "Play") INSTALLED_STATE_BUTTONS = ("Open", "Play", "Uninstall") DISMISS_BUTTONS = ("Continue", "Accept", "Allow", "Got it", "Skip", "No thanks", "Not now", "OK", "Done") # Aurora Store 系统安装对话框的包名 _PACKAGE_INSTALLER = "com.android.packageinstaller" ACCOUNT_BANNED_TEXTS = ( "Authentication is required. You need to sign in to your Google Account.", ) REGION_BLOCKED_TEXTS = ( "This item isn't available in your country.", ) NOT_FOUND_TEXTS = ( "Item not found", ) INCOMPATIBLE_TEXTS = ( "This app is available only for your other devices", "Your device isn't compatible with this version.", ) PAGE_LOAD_FAILED_TEXTS = ( "Try again", ) AURORA_FAILURE_TEXTS = ( "Session expired", "Token expired", "Too many requests", "Rate limit", "Not available", "App not found", ) # 系统安装失败弹窗中常见的错误文本 # 这类弹窗如果不关闭，会遮挡后续所有应用的安装界面 SYSTEM_INSTALL_ERROR_TEXTS = ( "App not installed", "Installation failed", "Can't install", "Install failed", "not installed", "Couldn't install", ) def start(self, package_name, max_retry=1, target_account=None, **kwargs): del max_retry, target_account, kwargs try: # Aurora Store 使用 Compose UI，需要关闭动画才能正常执行 uiautomator dump if self._store_type == STORE_AURORA: self._disable_animations() # 清理上一次残留的弹窗/状态，避免遮挡当前任务的 UI self._clear_screen() if self.is_installed(package_name): logger.info("%s download skipped, app already installed: %s", self._store_name, package_name) return True, "app already installed" opened, message = self._open_store_page(package_name) if not opened: return False, message return self._install_from_store(package_name) except Exception as exc: logger.error("%s download failed for %s: %s", self._store_name, package_name, exc) return False, str(exc) def stop(self): return None def is_installed(self, package_name: str) -> bool: result = self._adb.run(["shell", "pm", "path", package_name], check=False, timeout=10) if result.returncode != 0: return False return any(line.strip().startswith("package:") for line in (result.stdout or "").splitlines()) def _install_from_store(self, package_name: str) -> tuple[bool, str]: install_clicked = False system_install_clicked = False deadline = time.time() + self.INSTALL_TIMEOUT_SEC # 页面无进展计时：如果长时间没有出现 Install 按钮或安装状态， # 说明页面可能卡在加载界面（Aurora 找不到包名时的典型表现） no_progress_since: float | None = None # Aurora 点击 Install 后无反应计时： # 点击 Install 后按钮始终存在且无系统安装弹窗，说明 Aurora 未能触发下载 install_stuck_since: float | None = None while time.time() < deadline: if self.is_installed(package_name): return True, "install success" if install_clicked else "app already installed" nodes = self._dump_ui_nodes() if nodes: # ---- 失败检测 ---- failure = self._detect_failure(nodes) if failure is not None: # 尝试关闭错误弹窗，避免遗留弹窗遮挡后续任务的 UI self._dismiss_dialog(nodes) return False, failure # ---- 系统安装失败弹窗检测 ---- # 某些应用安装失败时，系统弹出 "App not installed" 等弹窗 # 如果不关闭，会遮挡后续所有应用的安装界面 sys_error = self._detect_system_install_error(nodes) if sys_error is not None: logger.warning("检测到系统安装失败弹窗: %s", sys_error) self._dismiss_dialog(nodes) return False, sys_error # ---- Aurora Store: 系统安装对话框确认 ---- # 仅在点击 Aurora 的 Install 按钮后才检测系统安装弹窗， # 避免识别到上一次下载遗留的弹窗导致 system_install_clicked 误置 if self._store_type == STORE_AURORA and install_clicked and not system_install_clicked: pkg_installer_node = self._find_node_by_package( nodes, self._PACKAGE_INSTALLER, ("Install",) ) if pkg_installer_node is not None: logger.info("Aurora: 检测到系统安装确认对话框，点击 INSTALL") if self._tap_node(pkg_installer_node): system_install_clicked = True no_progress_since = None time.sleep(5) continue # ---- 弹窗自动关闭 ---- if self._click_first(nodes, self.DISMISS_BUTTONS): no_progress_since = None time.sleep(2) continue # ---- 安装完成检测（双重校验） ---- if self._has_installed_state(nodes): # 双重校验：UI 显示已安装，再用 pm path 确认 # 防止 Aurora 安装过程中短暂出现 Open 按钮导致误判 if self.is_installed(package_name): return True, "install success" if install_clicked else "app already installed" logger.debug("UI 显示已安装但 pm path 未确认，继续等待: %s", package_name) # ---- 点击 Install/Update 按钮 ---- # 仅首次点击，避免重复点击时 continue 跳过下方的卡住检测 install_button = self._find_first(nodes, self.INSTALL_BUTTONS) if install_button is not None and not install_clicked: if self._tap_node(install_button): install_clicked = True no_progress_since = None logger.info("Clicked '%s' button in %s", install_button.text or install_button.desc, self._store_name) time.sleep(3) continue # ---- Aurora 点击 Install 后无反应检测 ---- # 点击 Install 后按钮始终存在、无系统安装弹窗 → Aurora 未能触发下载 if (install_clicked and install_button is not None and not system_install_clicked): if install_stuck_since is None: install_stuck_since = time.time() elif time.time() - install_stuck_since > self.AURORA_INSTALL_STUCK_SEC: logger.warning( "Aurora Install 点击后 %ds 无变化（按钮仍为'%s'），下载未触发: %s", self.AURORA_INSTALL_STUCK_SEC, install_button.text or install_button.desc, package_name, ) return False, "aurora install not triggered" else: install_stuck_since = None # ---- 加载超时检测 ---- # 判断是否有"有效进展"：有 Install/Update 按钮、安装状态、或失败信息 has_progress = ( install_clicked or install_button is not None or system_install_clicked ) if has_progress: no_progress_since = None else: if no_progress_since is None: no_progress_since = time.time() elif time.time() - no_progress_since > self.LOADING_TIMEOUT_SEC: logger.warning("%s 页面加载超时 (%ds)，包名可能不存在: %s", self._store_name, self.LOADING_TIMEOUT_SEC, package_name) return False, "page loading timeout - app may not exist" time.sleep(self.POLL_INTERVAL_SEC) return False, "install timeout" def _open_store_page(self, package_name: str) -> tuple[bool, str]: url = self._build_url(package_name) # 构建 intent 命令，通过 -p 参数显式指定目标商店包名 # 避免 market:// 被其他应用拦截（如设备上同时存在 Play Store 和 Aurora Store） am_cmd = [ "shell", "am", "start", "-a", "android.intent.action.VIEW", "-d", url, ] if self._store_package: am_cmd.extend(["-p", self._store_package]) result = self._adb.run(am_cmd, check=False, timeout=15) if result.returncode != 0: error = (result.stderr or result.stdout or "jump failed").strip() return False, error deadline = time.time() + self.OPEN_TIMEOUT_SEC while time.time() < deadline: if self._is_store_in_foreground(): return True, "opened" nodes = self._dump_ui_nodes() if nodes and ( self._detect_failure(nodes) is not None or self._find_first(nodes, self.INSTALL_BUTTONS) is not None or self._has_installed_state(nodes) ): return True, "opened" time.sleep(2) return False, "jump timeout" def _is_store_in_foreground(self) -> bool: current = get_current_package(lambda cmd: self._adb.shell(cmd, timeout=10)) return current == self._store_package def _ensure_single_device(self) -> None: if self._serial: return result = self._adb.run(["devices"], check=False, timeout=10) lines = [line.strip() for line in (result.stdout or "").splitlines()] online = [line for line in lines[1:] if line.endswith("\tdevice")] if len(online) != 1: raise RuntimeError(f"Expected exactly 1 online adb device, found {len(online)}.") def _is_google_play_in_foreground(self) -> bool: return self._is_store_in_foreground() def _dump_ui_nodes(self, max_retries: int = 3) -> list[UiNode]: """执行 uiautomator dump 并解析 UI 节点树。 Aurora Store 使用 Compose UI，偶尔会导致 dump 失败（ERROR: could not get idle state），因此增加重试逻辑。 """ for attempt in range(max_retries): dump_result = self._adb.run( ["shell", "uiautomator", "dump", "/sdcard/uidump.xml"], check=False, timeout=15, ) if dump_result.returncode != 0: dump_err = (dump_result.stdout or "").strip() if attempt < max_retries - 1: logger.debug("uiautomator dump 失败 (%d/%d): %s", attempt + 1, max_retries, dump_err) time.sleep(1) continue return [] read_result = self._adb.run( ["exec-out", "cat", "/sdcard/uidump.xml"], check=False, timeout=15, ) xml_text = (read_result.stdout or "").strip() if read_result.returncode != 0 or not xml_text: if attempt < max_retries - 1: time.sleep(1) continue return [] try: root = ET.fromstring(xml_text) except ET.ParseError: if attempt < max_retries - 1: time.sleep(1) continue return [] nodes: list[UiNode] = [] for raw in root.iter("node"): bounds = self._parse_bounds(raw.attrib.get("bounds", "")) if bounds is None: continue nodes.append( UiNode( text=str(raw.attrib.get("text", "") or ""), desc=str(raw.attrib.get("content-desc", "") or ""), package=str(raw.attrib.get("package", "") or ""), enabled=str(raw.attrib.get("enabled", "true")).lower() == "true", clickable=str(raw.attrib.get("clickable", "false")).lower() == "true", bounds=bounds, ) ) return nodes return [] def _detect_failure(self, nodes: list[UiNode]) -> str | None: labels = [label.lower() for node in nodes for label in node.labels()] if self._contains_text(labels, self.ACCOUNT_BANNED_TEXTS): return "account banned" if self._contains_text(labels, self.REGION_BLOCKED_TEXTS): return "app not available" if self._contains_text(labels, self.NOT_FOUND_TEXTS): return "app not found" if self._contains_text(labels, self.INCOMPATIBLE_TEXTS): return "app incompatible" if self._contains_text(labels, self.PAGE_LOAD_FAILED_TEXTS): return "app page load failed" if self._store_type == STORE_AURORA: if self._contains_text(labels, self.AURORA_FAILURE_TEXTS): return "aurora session/server error" return None def _detect_system_install_error(self, nodes: list[UiNode]) -> str | None: """检测系统安装失败弹窗。 某些应用安装失败时，Android 系统会弹出 "App not installed" 等错误对话框。 这类弹窗如果不关闭，会遮挡后续所有应用的安装界面，导致脚本无法获取 Install 按钮而误判为无法下载。 """ labels = [label.lower() for node in nodes for label in node.labels()] if self._contains_text(labels, self.SYSTEM_INSTALL_ERROR_TEXTS): return "system install error dialog" return None def _dismiss_dialog(self, nodes: list[UiNode] | None = None) -> None: """尝试关闭当前屏幕上的弹窗。 先尝试通过 UI 节点点击常见的关闭按钮（OK/Close/Got it 等）， 再按 Back 键作为兜底。 """ if nodes: dismiss_labels = ("OK", "Close", "Got it", "Done", "Cancel") self._click_first(nodes, dismiss_labels) time.sleep(1) # Back 键兜底关闭弹窗 self._adb.run(["shell", "input", "keyevent", "KEYCODE_BACK"], check=False, timeout=5) time.sleep(0.5) def _has_installed_state(self, nodes: list[UiNode]) -> bool: """判断应用是否已安装完成。 注意：点击 Install/Update 后，Google Play 页面可能会短暂显示灰色的 Open 按钮，此时如果同时存在 Cancel 按钮，说明安装仍在进行中。 必须以 Uninstall 按钮出现，或 Open/Play 且没有 Cancel/Install/Update 作为真正完成的标志。 """ # Uninstall 是最可靠的安装完成标志 if self._find_first(nodes, ("Uninstall",)) is not None: return True # 如果还有 Cancel 或 Install/Update 按钮，说明安装还在进行中 if self._find_first(nodes, ("Cancel", "Install", "Update")) is not None: return False # 只有 Open/Play 且没有上述安装相关按钮，算完成 if self._find_first(nodes, ("Open", "Play")) is not None: return True return False def _click_first(self, nodes: list[UiNode], labels: tuple[str, ...]) -> bool: node = self._find_first(nodes, labels) if node is None: return False return self._tap_node(node) def _find_first(self, nodes: list[UiNode], labels: tuple[str, ...]) -> UiNode | None: wanted = {label.strip().lower() for label in labels} matches: list[UiNode] = [] for node in nodes: if not node.enabled: continue for label in node.labels(): if label.strip().lower() in wanted: matches.append(node) break if not matches: return None return min(matches, key=lambda node: (node.center[1], node.center[0], not node.clickable)) def _find_node_by_package( self, nodes: list[UiNode], package: str, labels: tuple[str, ...], ) -> UiNode | None: """在指定包名的节点中查找匹配 label 的控件。 用于在 Aurora Store 场景中精确匹配系统安装对话框 (com.android.packageinstaller) 的按钮，避免与商店内同名按钮混淆。 """ wanted = {label.strip().lower() for label in labels} for node in nodes: if not node.enabled: continue if node.package != package: continue for label in node.labels(): if label.strip().lower() in wanted: return node return None def _disable_animations(self) -> None: """关闭设备动画。 Aurora Store 使用 Jetpack Compose 构建 UI，Compose 的动画/过渡效果 会导致 uiautomator dump 报 'ERROR: could not get idle state' 而失败。 关闭全局动画可以有效解决此问题。 """ for setting in ( "window_animation_scale", "transition_animation_scale", "animator_duration_scale", ): self._adb.run( ["shell", "settings", "put", "global", setting, "0"], check=False, timeout=10, ) logger.debug("已关闭设备动画 (serial=%s)", self._serial) def _clear_screen(self) -> None: """清理屏幕上可能残留的弹窗/遮挡。 在每次下载开始前调用，确保商店界面不被之前遗留的系统弹窗 （如安装失败提示）遮挡。 """ # 先回到主屏幕 self._adb.run(["shell", "input", "keyevent", "KEYCODE_HOME"], check=False, timeout=5) # 关闭商店应用（清除残留状态） self._adb.run(["shell", "am", "force-stop", self._store_package], check=False, timeout=5) # 关闭系统安装器（可能有残留的安装失败弹窗） self._adb.run(["shell", "am", "force-stop", self._PACKAGE_INSTALLER], check=False, timeout=5) time.sleep(0.5) def reset_state(self) -> None: """完成一个应用的下载/导出/卸载后重置设备状态。 在 us_download_worker 中每次处理完一个包后调用，确保： 1. 关闭残留的商店页面和弹窗 2. 回到主屏幕 3. 不影响下一个包的下载 """ self._adb.run(["shell", "input", "keyevent", "KEYCODE_HOME"], check=False, timeout=5) self._adb.run(["shell", "am", "force-stop", self._store_package], check=False, timeout=5) self._adb.run(["shell", "am", "force-stop", self._PACKAGE_INSTALLER], check=False, timeout=5) # 也关闭可能弹出的权限管理器等系统组件 self._adb.run(["shell", "am", "force-stop", "com.google.android.permissioncontroller"], check=False, timeout=5) time.sleep(0.5) logger.debug("设备状态已重置 (serial=%s)", self._serial) def _tap_node(self, node: UiNode) -> bool: x, y = node.center if x <= 0 and y <= 0: return False result = self._adb.run( ["shell", "input", "tap", str(x), str(y)], check=False, timeout=10, ) return result.returncode == 0 @staticmethod def _contains_text(labels: list[str], expected_values: tuple[str, ...]) -> bool: for expected in expected_values: expected_lower = expected.lower() if any(expected_lower in label for label in labels): return True return False @staticmethod def _parse_bounds(value: str) -> tuple[int, int, int, int] | None: match = _BOUNDS_RE.fullmatch(str(value or "").strip()) if match is None: return None left, top, right, bottom = (int(part) for part in match.groups()) return left, top, right, bottom def export_apk(self, package_name: str, output_dir: str) -> list: os.makedirs(output_dir, exist_ok=True) result = self._adb.run( ["shell", "pm", "path", package_name], check=False, timeout=10, ) if result.returncode != 0: logger.warning("pm path failed for %s: %s", package_name, (result.stderr or "").strip()) return [] apk_paths = [] for line in (result.stdout or "").splitlines(): line = line.strip() if line.startswith("package:"): apk_paths.append(line.split(":", 1)[1].strip()) if not apk_paths: logger.warning("No APK paths found for %s", package_name) return [] exported = [] for apk_path in apk_paths: filename = os.path.basename(apk_path) dest = os.path.join(output_dir, filename) pull_result = self._adb.run( ["pull", apk_path, dest], check=False, timeout=60, ) if pull_result.returncode != 0 or not os.path.isfile(dest): logger.warning("adb pull failed for %s: %s", package_name, apk_path) continue exported.append(dest) return exported def get_apk_version(self, package_name: str) -> str: result = self._adb.run( ["shell", "dumpsys", "package", package_name], check=False, timeout=15, ) if result.returncode != 0: return "" for line in (result.stdout or "").splitlines(): if "versionName=" in line: val = line.split("versionName=", 1)[1].strip() if val: return val return "" @staticmethod def _build_google_play_url(package_name: str) -> str: return f"https://play.google.com/store/apps/details?id={package_name}" @staticmethod def _build_aurora_url(package_name: str) -> str: return f"market://details?id={package_name}" def main() -> int: import argparse parser = argparse.ArgumentParser(description="Google Play / Aurora Store Downloader") parser.add_argument("--package", default="com.google.android.youtube", help="Package name to download") parser.add_argument("--serial", default="", help="Target device serial (optional)") parser.add_argument("--aurora", action="store_true", help="Use Aurora Store instead of Google Play") args = parser.parse_args() logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) store_type = STORE_AURORA if args.aurora else STORE_GOOGLE_PLAY downloader = GooglePlayDownloader( serial=args.serial or None, store_type=store_type, ) success, message = downloader.start(args.package) print(f"success={success}") print(f"message={message}") return 0 if success else 1 if __name__ == "__main__": raise SystemExit(main())