665 lines
26 KiB
Python
665 lines
26 KiB
Python
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())
|