649 lines
25 KiB
Python
649 lines
25 KiB
Python
# -*- encoding=utf8 -*-
|
|
import os
|
|
import sys
|
|
import time
|
|
import logging
|
|
from typing import Tuple
|
|
from airtest.core.api import *
|
|
from airtest.core.android import *
|
|
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
|
|
from poco.exceptions import PocoTargetTimeout, PocoNoSuchNodeException
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, sys.path[0].rsplit('\\', 3)[0])
|
|
from adb_helper import ADBHelper
|
|
from config_loader import load_config as load_autool_config
|
|
from result_codes import DownloadError
|
|
from utils_android.device_config import build_airtest_android_uri, ensure_android_wireless_connected, get_android_device_serial
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class GooglePlayDownloader:
|
|
GOOGLE_PLAY_PACKAGE = "com.android.vending"
|
|
APP_INCOMPATIBLE_TEXTS = (
|
|
"This app is available only for your other devices",
|
|
"Your device isn't compatible with this version.",
|
|
)
|
|
ITEM_NOT_FOUND_TEXT = "Item not found"
|
|
ITEM_NOT_FOUND_IMAGE = "item_not_found.png"
|
|
ITEM_NOT_FOUND_IMAGE_THRESHOLD = 0.7
|
|
TRY_AGAIN_TEXT = "Try again"
|
|
GOOGLE_PLAY_WEB_NOT_FOUND_PATTERNS = (
|
|
"item not found",
|
|
"requested url was not found",
|
|
"the requested url was not found",
|
|
"url was not found on this server",
|
|
)
|
|
PRIMARY_ACTION_TEXTS = ("Install", "Update", "Open", "Play")
|
|
GOOGLE_PLAY_PRECHECK_TIMEOUT = 60
|
|
GOOGLE_PLAY_PRECHECK_REQUEST_TIMEOUT = 10
|
|
GOOGLE_PLAY_PRECHECK_RETRY_INTERVAL = 2
|
|
CURL_HTTP_CODE_MARKER = "__CURL_HTTP_CODE__:"
|
|
GOOGLE_PLAY_OPEN_TIMEOUT = 60
|
|
GOOGLE_PLAY_ACTION_TIMEOUT = 30
|
|
|
|
ERROR_MAP = {
|
|
'app not available': DownloadError.REGION_RESTRICTED,
|
|
'account banned': DownloadError.ACCOUNT_BANNED,
|
|
'app not found': DownloadError.APP_NOT_FOUND,
|
|
'app incompatible': DownloadError.INCOMPATIBLE,
|
|
'install overtime': DownloadError.DOWNLOAD_TIMEOUT,
|
|
'jump timeout': DownloadError.DOWNLOAD_TIMEOUT,
|
|
'timeout': DownloadError.DOWNLOAD_TIMEOUT,
|
|
'jump failed': DownloadError.NETWORK_ERROR,
|
|
'app page load failed': DownloadError.NETWORK_ERROR,
|
|
'click download button failed': DownloadError.INSTALL_FAILED,
|
|
'install success': None,
|
|
'app already installed': None,
|
|
}
|
|
|
|
def __init__(self):
|
|
self.poco = None
|
|
self.adb_helper = None
|
|
self._device_curl_available = None
|
|
|
|
def _map_error(self, state: str) -> Tuple[bool, DownloadError]:
|
|
"""将字符串状态映射为 DownloadError"""
|
|
if state in ('install success', 'app already installed'):
|
|
return True, None
|
|
error = self.ERROR_MAP.get(state, DownloadError.OTHER)
|
|
return False, error
|
|
|
|
def _ensure_poco(self):
|
|
if self.poco is None:
|
|
try:
|
|
config = load_autool_config()
|
|
serial = get_android_device_serial(config)
|
|
ensure_android_wireless_connected(config, serial)
|
|
auto_setup(
|
|
__file__,
|
|
devices=[
|
|
build_airtest_android_uri(
|
|
config,
|
|
serial=serial,
|
|
cap_method="ADBCAP",
|
|
ori_method="ADBORI",
|
|
)
|
|
]
|
|
)
|
|
|
|
# Start poco service if needed
|
|
start_app("com.netease.open.pocoservice")
|
|
sleep(2)
|
|
self.poco = AndroidUiautomationPoco(use_airtest_input=True, screenshot_each_action=False)
|
|
except Exception as e:
|
|
logger.error(f"Poco initialization failed: {e}")
|
|
|
|
|
|
def _ensure_adb_helper(self):
|
|
if self.adb_helper is None:
|
|
self.adb_helper = ADBHelper()
|
|
return self.adb_helper
|
|
|
|
def is_account_banned(self):
|
|
"""检测账号是否被封禁"""
|
|
ban_texts = [
|
|
"Authentication is required. You need to sign in to your Google Account.",
|
|
]
|
|
for text in ban_texts:
|
|
if self.poco(text=text).exists():
|
|
logger.warning(f"APP Downloader: Account ban detected - '{text}'")
|
|
return True
|
|
return False
|
|
|
|
def is_app_incompatible(self, package=None):
|
|
"""检测应用是否不兼容当前设备"""
|
|
for incompatible_text in self.APP_INCOMPATIBLE_TEXTS:
|
|
query = {"text": incompatible_text}
|
|
if package:
|
|
query["package"] = package
|
|
if self.poco(**query).exists():
|
|
logger.warning(
|
|
"APP Downloader: App incompatible with current device: '%s'",
|
|
incompatible_text,
|
|
)
|
|
return True
|
|
return False
|
|
|
|
def is_try_again_page_load_failed(self, package=None):
|
|
query = {"text": self.TRY_AGAIN_TEXT}
|
|
if package:
|
|
query["package"] = package
|
|
if self.poco(**query).exists():
|
|
logger.warning("APP Downloader: Try again detected, treating as page load failed")
|
|
return True
|
|
return False
|
|
|
|
def is_app_not_available_by_error_image(self):
|
|
picture_path = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)),
|
|
self.ITEM_NOT_FOUND_IMAGE,
|
|
)
|
|
tpl = Template(
|
|
picture_path,
|
|
threshold=self.ITEM_NOT_FOUND_IMAGE_THRESHOLD,
|
|
rgb=True,
|
|
)
|
|
if exists(tpl):
|
|
logger.warning(
|
|
"APP Downloader: Google Play region-restricted error illustration detected"
|
|
)
|
|
return True
|
|
return False
|
|
|
|
def is_item_not_found(self, package=None):
|
|
query = {"text": self.ITEM_NOT_FOUND_TEXT}
|
|
if package:
|
|
query["package"] = package
|
|
if self.poco(**query).exists():
|
|
logger.warning("APP Downloader: Item not found detected")
|
|
return True
|
|
return False
|
|
|
|
def _get_control_y_position(self, control):
|
|
try:
|
|
position = control.get_position()
|
|
except Exception:
|
|
return None
|
|
|
|
if not position or len(position) < 2:
|
|
return None
|
|
return position[1]
|
|
|
|
def _build_google_play_url(self, package_name, country=None):
|
|
url = f"https://play.google.com/store/apps/details?id={package_name}&hl=en"
|
|
if country:
|
|
url += f"&gl={country}"
|
|
return url
|
|
|
|
def _run_device_curl_precheck(self, url):
|
|
output = self._ensure_adb_helper().shell(
|
|
[
|
|
"curl",
|
|
"-L",
|
|
"--max-time",
|
|
str(self.GOOGLE_PLAY_PRECHECK_REQUEST_TIMEOUT),
|
|
"-A",
|
|
(
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/124.0.0.0 Safari/537.36"
|
|
),
|
|
"-H",
|
|
"Accept-Language: en-US,en;q=0.9",
|
|
"-sS",
|
|
url,
|
|
"-w",
|
|
f"\n{self.CURL_HTTP_CODE_MARKER}%{{http_code}}\n",
|
|
]
|
|
)
|
|
return str(output or "")
|
|
|
|
def _has_device_curl(self):
|
|
if self._device_curl_available is not None:
|
|
return self._device_curl_available
|
|
|
|
output = self._ensure_adb_helper().shell(["sh", "-c", "command -v curl || which curl || true"])
|
|
self._device_curl_available = bool(str(output or "").strip())
|
|
if not self._device_curl_available:
|
|
logger.warning(
|
|
"APP Downloader: Device has no curl; skipping Google Play webpage precheck"
|
|
)
|
|
return self._device_curl_available
|
|
|
|
def _parse_device_curl_output(self, output):
|
|
http_code = ""
|
|
body_lines = []
|
|
for line in str(output or "").splitlines():
|
|
if line.startswith(self.CURL_HTTP_CODE_MARKER):
|
|
http_code = line[len(self.CURL_HTTP_CODE_MARKER):].strip()
|
|
continue
|
|
body_lines.append(line)
|
|
return http_code, "\n".join(body_lines)
|
|
|
|
def precheck_google_play_webpage(
|
|
self,
|
|
package_name,
|
|
country=None,
|
|
timeout=GOOGLE_PLAY_PRECHECK_TIMEOUT,
|
|
):
|
|
"""在拉起浏览器/Google Play 前，使用设备侧 curl 阻塞式检查网页是否可访问或为 not found 页面"""
|
|
url = self._build_google_play_url(package_name, country)
|
|
|
|
if not self._has_device_curl():
|
|
return True, "precheck skipped: device curl unavailable"
|
|
|
|
deadline = time.time() + timeout
|
|
last_error = "device curl precheck returned empty output"
|
|
while time.time() < deadline:
|
|
try:
|
|
logger.info(f"APP Downloader: Prechecking Google Play webpage {url}")
|
|
output = self._run_device_curl_precheck(url)
|
|
except Exception as exc:
|
|
last_error = exc
|
|
logger.warning(
|
|
"APP Downloader: Device-side Google Play precheck attempt failed, will retry until timeout: %s",
|
|
exc,
|
|
)
|
|
if time.time() >= deadline:
|
|
break
|
|
sleep(self.GOOGLE_PLAY_PRECHECK_RETRY_INTERVAL)
|
|
continue
|
|
|
|
http_code, page_text = self._parse_device_curl_output(output)
|
|
page_text_lower = page_text.lower()
|
|
if http_code == "404" or any(
|
|
pattern in page_text_lower for pattern in self.GOOGLE_PLAY_WEB_NOT_FOUND_PATTERNS
|
|
):
|
|
logger.info("APP Downloader: Google Play webpage reports app not found")
|
|
return False, "app not found"
|
|
|
|
if http_code:
|
|
logger.info(
|
|
"APP Downloader: Device-side Google Play webpage precheck passed with status %s",
|
|
http_code or "200",
|
|
)
|
|
return True, "precheck ok"
|
|
|
|
last_error = output or "device curl returned no HTTP status marker"
|
|
logger.warning(
|
|
"APP Downloader: Device-side Google Play precheck attempt failed with http=%s; "
|
|
"will retry until timeout",
|
|
http_code or "unknown",
|
|
)
|
|
if time.time() >= deadline:
|
|
break
|
|
sleep(self.GOOGLE_PLAY_PRECHECK_RETRY_INTERVAL)
|
|
|
|
logger.error(
|
|
"APP Downloader: Device-side Google Play webpage precheck timed out after %ss: %s",
|
|
timeout,
|
|
last_error,
|
|
)
|
|
return False, "timeout"
|
|
|
|
def _get_primary_action_button(self, action_names=None):
|
|
best_candidate = None
|
|
best_sort_key = None
|
|
|
|
for action_index, action_name in enumerate(action_names or self.PRIMARY_ACTION_TEXTS):
|
|
controls = self.poco(text=action_name, package=self.GOOGLE_PLAY_PACKAGE)
|
|
if not controls.exists():
|
|
continue
|
|
|
|
try:
|
|
candidates = list(controls)
|
|
except TypeError:
|
|
candidates = [controls]
|
|
|
|
for control in candidates:
|
|
y_pos = self._get_control_y_position(control)
|
|
sort_key = (
|
|
y_pos if y_pos is not None else float("inf"),
|
|
action_index,
|
|
)
|
|
if best_sort_key is None or sort_key < best_sort_key:
|
|
best_sort_key = sort_key
|
|
best_candidate = (action_name, control)
|
|
|
|
if best_candidate is not None:
|
|
return best_candidate
|
|
return None, None
|
|
|
|
def _wait_for_google_play_ready(
|
|
self,
|
|
timeout=GOOGLE_PLAY_ACTION_TIMEOUT,
|
|
):
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if self.poco(text="This item isn't available in your country.", package=self.GOOGLE_PLAY_PACKAGE).exists():
|
|
logger.info("APP Downloader: This item isn't available in your country!")
|
|
return False, 'app not available'
|
|
|
|
if self.is_app_not_available_by_error_image():
|
|
return False, 'app not available'
|
|
|
|
if self.is_item_not_found(package=self.GOOGLE_PLAY_PACKAGE) or self.is_item_not_found():
|
|
return False, 'app not found'
|
|
|
|
if self.is_try_again_page_load_failed(package=self.GOOGLE_PLAY_PACKAGE) or self.is_try_again_page_load_failed():
|
|
return False, 'app page load failed'
|
|
|
|
if self.is_app_incompatible(package=self.GOOGLE_PLAY_PACKAGE) or self.is_app_incompatible():
|
|
return False, 'app incompatible'
|
|
|
|
action_name, action_button = self._get_primary_action_button()
|
|
if action_button is not None:
|
|
logger.info(
|
|
"APP Downloader: Google Play detail page ready with %s text",
|
|
action_name,
|
|
)
|
|
return True, action_name
|
|
|
|
if self.is_account_banned():
|
|
return False, 'account banned'
|
|
|
|
sleep(1)
|
|
|
|
logger.warning(
|
|
"APP Downloader: Google Play opened but no primary action text appeared before timeout"
|
|
)
|
|
return False, 'app page load failed'
|
|
|
|
def click_download_button(self):
|
|
"""点击下载按钮，只处理下载逻辑，不处理账号切换"""
|
|
# Check regional availability
|
|
if self.poco(text="This item isn't available in your country.", package=self.GOOGLE_PLAY_PACKAGE).exists():
|
|
logger.info("APP Downloader: This item isn't available in your country!")
|
|
return False, 'app not available'
|
|
|
|
if self.is_app_not_available_by_error_image():
|
|
return False, 'app not available'
|
|
|
|
# Check if account is banned
|
|
if self.is_account_banned():
|
|
logger.warning("APP Downloader: Account banned")
|
|
return False, 'account banned'
|
|
|
|
# Wait for buttons
|
|
try:
|
|
logger.info("APP Downloader: Waiting for app page to load...")
|
|
ready, state = self._wait_for_google_play_ready(
|
|
timeout=self.GOOGLE_PLAY_ACTION_TIMEOUT,
|
|
)
|
|
if not ready:
|
|
return False, state
|
|
|
|
if self.is_app_incompatible(package=self.GOOGLE_PLAY_PACKAGE) or self.is_app_incompatible():
|
|
return False, 'app incompatible'
|
|
|
|
action_name, action_button = self._get_primary_action_button(("Open", "Play"))
|
|
if action_button is not None or self.poco(text="Uninstall", package=self.GOOGLE_PLAY_PACKAGE).exists():
|
|
if self.is_account_banned():
|
|
logger.warning("APP Downloader: Account banned before installed-state success")
|
|
return False, 'account banned'
|
|
return True, 'app already installed'
|
|
|
|
action_name, action_button = self._get_primary_action_button(("Update", "Install"))
|
|
if action_button is None:
|
|
return False, 'click download button failed'
|
|
|
|
if action_name == "Update":
|
|
logger.info("APP Downloader: Found Update button, clicking...")
|
|
action_button.click()
|
|
sleep(3)
|
|
self.deal_exception()
|
|
return self.wait_for_download()
|
|
|
|
if action_name == "Install":
|
|
action_button.click()
|
|
sleep(3)
|
|
self.deal_exception()
|
|
return self.wait_for_download()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking buttons: {e}")
|
|
return False, 'app page load failed'
|
|
|
|
return False, 'click download button failed'
|
|
|
|
def deal_exception(self):
|
|
"""Handle common popups"""
|
|
ins_text_list = ['Skip','No thanks','Not now','Continue','Accept','Got it','No','Yes, it\'s me']
|
|
for ins_text in ins_text_list:
|
|
if self.poco(text=ins_text).exists():
|
|
logger.info(f"APP Downloader: Handling popup: {ins_text}")
|
|
self.poco(text=ins_text).click()
|
|
sleep(3)
|
|
|
|
def wait_for_download(self):
|
|
"""Wait for the download to complete"""
|
|
retry_times_max = 60 # 10 minutes
|
|
for _ in range(retry_times_max):
|
|
# Check if still downloading (Cancel button exists)
|
|
if self.poco(name='Cancel', package=self.GOOGLE_PLAY_PACKAGE).exists():
|
|
logger.info("APP Downloader: Still downloading...")
|
|
sleep(10)
|
|
continue
|
|
|
|
if self.is_app_not_available_by_error_image():
|
|
return False, 'app not available'
|
|
|
|
if self.is_item_not_found(package=self.GOOGLE_PLAY_PACKAGE) or self.is_item_not_found():
|
|
return False, 'app not found'
|
|
|
|
if self.is_try_again_page_load_failed(package=self.GOOGLE_PLAY_PACKAGE) or self.is_try_again_page_load_failed():
|
|
return False, 'app page load failed'
|
|
|
|
if self.is_app_incompatible(package=self.GOOGLE_PLAY_PACKAGE) or self.is_app_incompatible():
|
|
return False, 'app incompatible'
|
|
|
|
# Check for success indicators
|
|
action_name, action_button = self._get_primary_action_button(("Open", "Play"))
|
|
if action_button is not None:
|
|
if self.is_account_banned():
|
|
logger.warning("APP Downloader: Account banned before download-complete success")
|
|
return False, 'account banned'
|
|
logger.info("APP Downloader: Download complete")
|
|
return True, 'install success'
|
|
|
|
sleep(10)
|
|
self.deal_exception()
|
|
|
|
return False, 'install overtime'
|
|
|
|
def intent_jump_to_google_play(self, package_name, country=None):
|
|
"""Open Google Play via Intent"""
|
|
url = self._build_google_play_url(package_name, country)
|
|
logger.info(f"APP Downloader: Opening Google Play via intent: {url}")
|
|
|
|
shell(f'am start -a android.intent.action.VIEW -d "{url}"')
|
|
sleep(3)
|
|
|
|
deadline = time.time() + self.GOOGLE_PLAY_OPEN_TIMEOUT
|
|
while time.time() < deadline:
|
|
top_app = shell("dumpsys window | grep mCurrentFocus")
|
|
if self.GOOGLE_PLAY_PACKAGE in str(top_app):
|
|
logger.info("APP Downloader: Switched to Google Play")
|
|
remaining_timeout = max(1, deadline - time.time())
|
|
return self._wait_for_google_play_ready(
|
|
timeout=remaining_timeout,
|
|
)
|
|
|
|
sleep(1)
|
|
|
|
return False, "jump timeout"
|
|
|
|
def start(self, package_name, country=None, max_retry=3, target_account=None, **kwargs):
|
|
"""
|
|
Main entry point for Google Play Downloader
|
|
:param max_retry: 最大重试次数（账号被封时切换账号重试）
|
|
:param target_account: 目标账号邮箱（可选），如果提供则切换到指定账号
|
|
"""
|
|
self._ensure_poco()
|
|
logger.info(f"Starting Google Play download for {package_name}")
|
|
|
|
retry_count = 0
|
|
while retry_count < max_retry:
|
|
# 1. 打开 Google Play 页面
|
|
jump_result = self.intent_jump_to_google_play(package_name, country)
|
|
if isinstance(jump_result, tuple):
|
|
success, error_state = jump_result
|
|
if not success:
|
|
logger.error(f"Failed to jump to Google Play: {error_state}")
|
|
return False, error_state
|
|
|
|
# 2. 点击下载
|
|
success, state = self.click_download_button()
|
|
|
|
# 3. 处理结果
|
|
if success:
|
|
# 下载成功
|
|
stop_app(self.GOOGLE_PLAY_PACKAGE)
|
|
return True, state
|
|
|
|
if state == 'account banned':
|
|
# 账号被封，尝试切换
|
|
logger.warning(f"APP Downloader: Account banned, switching account (retry {retry_count + 1}/{max_retry})")
|
|
switch_success, switch_msg = self.switch_google_account(target_account=target_account)
|
|
if not switch_success:
|
|
logger.error(f"APP Downloader: Failed to switch account: {switch_msg}")
|
|
stop_app(self.GOOGLE_PLAY_PACKAGE)
|
|
return False, f'account banned, switch failed: {switch_msg}'
|
|
|
|
# 切换成功，关闭应用准备重试
|
|
logger.info("APP Downloader: Account switched, will retry...")
|
|
stop_app(self.GOOGLE_PLAY_PACKAGE)
|
|
sleep(2)
|
|
retry_count += 1
|
|
continue
|
|
|
|
# 其他错误（非账号问题），直接返回
|
|
stop_app(self.GOOGLE_PLAY_PACKAGE)
|
|
return False, state
|
|
|
|
# 超过最大重试次数
|
|
stop_app(self.GOOGLE_PLAY_PACKAGE)
|
|
return False, 'max retry exceeded'
|
|
|
|
def switch_google_account(self, target_account=None):
|
|
"""
|
|
切换谷歌账号
|
|
:param target_account: 目标账号邮箱（可选），如果提供则切换到指定账号，否则随机选择一个
|
|
:return: (bool, str) - (是否成功, 状态信息)
|
|
"""
|
|
try:
|
|
self._ensure_poco()
|
|
logger.info("APP Downloader: Starting Google account switch...")
|
|
|
|
# 1. 点击右上角账号头像按钮
|
|
touch((486, 42))
|
|
sleep(3)
|
|
|
|
# 2. 点击 "Switch account"
|
|
if not self.poco(text="Switch account").exists():
|
|
return False, "Switch account button not found"
|
|
self.poco(text="Switch account").click()
|
|
sleep(2)
|
|
|
|
# 3. 获取所有账号
|
|
accounts = self.poco(textMatches=".*@gmail\\.com.*")
|
|
if not accounts.exists():
|
|
return False, "No accounts found"
|
|
|
|
account_list = accounts.get_text()
|
|
if isinstance(account_list, str):
|
|
account_list = [account_list]
|
|
logger.info(f"APP Downloader: Available accounts: {account_list}")
|
|
|
|
if len(account_list) == 0:
|
|
return False, "Empty account list"
|
|
|
|
# 4. 选择账号
|
|
if target_account:
|
|
# 指定了账号：匹配并点击
|
|
target = None
|
|
for acc in account_list:
|
|
if target_account.lower() in acc.lower():
|
|
target = acc
|
|
break
|
|
if not target:
|
|
return False, f"Account '{target_account}' not found"
|
|
else:
|
|
# 未指定：随机选择（排除第一个）
|
|
import random
|
|
if len(account_list) > 1:
|
|
target = random.choice(account_list[1:])
|
|
else:
|
|
target = account_list[0]
|
|
|
|
# 5. 点击选中的账号
|
|
self.poco(text=target).click()
|
|
logger.info(f"APP Downloader: Switched to {target}")
|
|
sleep(3)
|
|
return True, f"switched to {target}"
|
|
|
|
except Exception as e:
|
|
logger.error(f"APP Downloader: Account switch failed: {e}")
|
|
return False, f"account switch failed: {str(e)}"
|
|
|
|
def dump_ui_tree(self, save_path=None):
|
|
"""
|
|
打印当前所有 UI 控件，用于调试
|
|
:param save_path: 可选，保存到文件路径
|
|
"""
|
|
try:
|
|
self._ensure_poco()
|
|
logger.info("=" * 60)
|
|
logger.info("APP Downloader: Dumping UI tree...")
|
|
logger.info("=" * 60)
|
|
|
|
# 获取所有控件
|
|
all_nodes = self.poco()
|
|
|
|
output_lines = []
|
|
for node in all_nodes:
|
|
try:
|
|
# 获取控件属性 - attr() 需要传入属性名
|
|
name = node.attr('name') or ''
|
|
text = node.attr('text') or ''
|
|
desc = node.attr('content-desc') or node.attr('desc') or ''
|
|
node_type = node.attr('type') or ''
|
|
pos = node.attr('pos') or ''
|
|
|
|
line = f"[{node_type}] name={name}, text='{text}', desc='{desc}', pos={pos}"
|
|
output_lines.append(line)
|
|
logger.info(line)
|
|
except Exception as node_e:
|
|
logger.warning(f"APP Downloader: Failed to get node info: {node_e}")
|
|
|
|
logger.info("=" * 60)
|
|
logger.info(f"APP Downloader: Total nodes: {len(output_lines)}")
|
|
logger.info("=" * 60)
|
|
|
|
# 保存到文件
|
|
if save_path:
|
|
with open(save_path, 'w', encoding='utf-8') as f:
|
|
f.write('\n'.join(output_lines))
|
|
logger.info(f"APP Downloader: UI tree saved to {save_path}")
|
|
|
|
return output_lines
|
|
|
|
except Exception as e:
|
|
logger.error(f"APP Downloader: Failed to dump UI tree: {e}")
|
|
return []
|
|
|
|
def stop(self):
|
|
"""Stop the poco instance"""
|
|
if self.poco:
|
|
try:
|
|
self.poco.stop_running()
|
|
logger.info("Google Play Downloader: Poco stopped")
|
|
except Exception as e:
|
|
logger.error(f"Error stopping poco: {e}")
|
|
finally:
|
|
self.poco = None
|
|
|
|
if __name__ == "__main__":
|
|
pass
|