# -*- encoding=utf8 -*- """ Download Controller Unified entry point for downloading/installing apps from various sources. """ import logging import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)).rsplit('\\', 3)[0]) from config_loader import load_config as load_autool_config from utils_android.device_config import build_airtest_android_uri, ensure_android_wireless_connected, get_android_device_serial from google_play_downloader import GooglePlayDownloader from apkpure_downloader import ApkPureDownloader from local_file_importer import LocalFileImporter import apks_export_import from result_codes import DownloadError from country_codes import normalize_country_codes logging.getLogger("airtest").setLevel(logging.ERROR) logging.getLogger("adb").setLevel(logging.ERROR) logging.getLogger("airtest.core.android.adb").setLevel(logging.ERROR) logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) _downloaders = {} _config_cache = None SUPPORTED_SOURCES = ('google_play', 'local') DEFAULT_SOURCES = ['google_play', 'local'] GOOGLE_PLAY_COUNTRY_CONTINUE_ERRORS = { DownloadError.REGION_RESTRICTED, DownloadError.APP_NOT_FOUND, DownloadError.INCOMPATIBLE, } LOCAL_FALLBACK_ELIGIBLE_ERRORS = GOOGLE_PLAY_COUNTRY_CONTINUE_ERRORS LOCAL_SOURCE_UNAVAILABLE_ERRORS = { "folder not found", "base.apk not found and xapk missing", "no apk files", "invalid xapk", "xapk missing apk files", } def _load_autool_config(): global _config_cache if _config_cache is not None: return _config_cache _config_cache = load_autool_config() return _config_cache DOWNLOAD_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 failed': DownloadError.INSTALL_FAILED, 'download failed': DownloadError.DOWNLOAD_TIMEOUT, 'install not verified': DownloadError.INSTALL_FAILED, 'network error': DownloadError.NETWORK_ERROR, 'source unavailable': DownloadError.SOURCE_UNAVAILABLE, 'folder not found': DownloadError.SOURCE_UNAVAILABLE, 'base.apk not found and xapk missing': DownloadError.SOURCE_UNAVAILABLE, 'no apk files': DownloadError.SOURCE_UNAVAILABLE, 'invalid xapk': DownloadError.SOURCE_UNAVAILABLE, 'xapk missing apk files': DownloadError.SOURCE_UNAVAILABLE, } def _parse_download_error(message: str, source: str = None) -> DownloadError: """将错误消息字符串转换为 DownloadError""" message_lower = str(message or '').strip().lower() exact_match = DOWNLOAD_ERROR_MAP.get(message_lower) if exact_match is not None: return exact_match if str(source or '').strip().lower() == 'local' and message_lower in LOCAL_SOURCE_UNAVAILABLE_ERRORS: return DownloadError.SOURCE_UNAVAILABLE if 'region' in message_lower or 'not available in your country' in message_lower or 'country restricted' in message_lower: return DownloadError.REGION_RESTRICTED if 'banned' in message_lower or 'authentication' in message_lower: return DownloadError.ACCOUNT_BANNED if 'not found' in message_lower: return DownloadError.APP_NOT_FOUND if 'incompatible' in message_lower or 'other devices' in message_lower or '不兼容' in message_lower or '不支持' in message_lower: return DownloadError.INCOMPATIBLE if 'timeout' in message_lower or 'overtime' in message_lower: return DownloadError.DOWNLOAD_TIMEOUT if 'network' in message_lower or 'connection' in message_lower: return DownloadError.NETWORK_ERROR if 'install' in message_lower: return DownloadError.INSTALL_FAILED return DownloadError.OTHER def _run_state_to_error(run_state: str): if run_state == 'not_installed': return DownloadError.INSTALL_FAILED, 'install not verified' if run_state == 'unknown': return DownloadError.OTHER, 'app state check returned unknown' return DownloadError.OTHER, f'unexpected run state ({run_state})' def _build_google_play_error_key(country): return f"google_play:{country}" def _normalize_available_sources(available_sources): if isinstance(available_sources, str): requested_sources = [available_sources] else: requested_sources = available_sources or DEFAULT_SOURCES normalized = [] seen = set() for source in requested_sources: source_name = str(source or '').strip() if not source_name or source_name not in SUPPORTED_SOURCES or source_name in seen: continue seen.add(source_name) normalized.append(source_name) if not normalized: normalized = list(DEFAULT_SOURCES) return normalized def _build_errors_json(errors): errors_json = {} for item in errors: entry = { 'code': item['code'], 'message': item['message'], } if item.get('run') is not None: entry['run'] = item['run'] errors_json[item['source']] = entry return errors_json def _build_failed_result(package_name, primary_error, errors, attempted_countries=None): result = { 'package_name': package_name, 'source': primary_error['source'], 'state': 'failed', 'run': primary_error.get('run', 'unknown'), 'error_code': primary_error['code'], 'details': primary_error['message'], 'errors': _build_errors_json(errors), } if attempted_countries: result['attempted_countries'] = attempted_countries return result def _can_fallback_to_local(country_errors): return bool(country_errors) and all( item['code'] in LOCAL_FALLBACK_ELIGIBLE_ERRORS for item in country_errors ) def _is_country_locked_terminal(errors): if not errors: return False google_play_codes = [] local_codes = [] for item in errors: if not isinstance(item, dict): return False source_name = str(item.get('source') or '').strip().lower() code = item.get('code') if source_name.startswith('google_play:'): google_play_codes.append(code) elif source_name == 'local': local_codes.append(code) if not google_play_codes or not local_codes: return False if any(code not in GOOGLE_PLAY_COUNTRY_CONTINUE_ERRORS for code in google_play_codes): return False return True def _pick_terminal_download_primary_error(errors): priority_by_code = { DownloadError.APP_NOT_FOUND: 1, DownloadError.REGION_RESTRICTED: 2, DownloadError.INCOMPATIBLE: 3, } best_item = None best_priority = -1 for item in errors or []: source_name = str(item.get('source') or '').strip().lower() code = item.get('code') if not source_name.startswith('google_play:') or code not in GOOGLE_PLAY_COUNTRY_CONTINUE_ERRORS: continue priority = priority_by_code.get(code, 0) if priority > best_priority: best_item = item best_priority = priority if best_item is not None: return best_item return (errors or [])[-1] if errors else { 'source': 'unknown', 'code': DownloadError.ALL_SOURCES_FAILED, 'message': 'All sources failed', 'run': 'unknown', } def _get_downloader(platform): """ Factory method to get or create downloader instances. """ if platform not in _downloaders: if platform == 'google_play': _downloaders[platform] = GooglePlayDownloader() elif platform == 'apkpure': _downloaders[platform] = ApkPureDownloader() elif platform == 'local': _downloaders[platform] = LocalFileImporter() return _downloaders.get(platform) import time from airtest.core.api import * # 导入 ADB Helper 和异常 from adb_helper import ADBHelper from DroidBot.exceptions import ADBException # 延迟初始化的 ADB Helper 实例 _adb_helper = None def _get_adb_helper(): """获取 ADB Helper 实例（延迟初始化） 避免在模块导入时就尝试连接设备，这样可以让主进程有机会 先恢复模拟器，再执行 ADB 相关操作。 """ global _adb_helper if _adb_helper is None: _adb_helper = ADBHelper() return _adb_helper def check_app_state(package_name): """ Check if the app runs normally or crashes. Returns: 'normal', 'crash', 'not_installed', or 'unknown' """ app_state = 'unknown' try: logger.info(f"Checking app state for {package_name}...") config = _load_autool_config() serial = get_android_device_serial(config) ensure_android_wireless_connected(config, serial) auto_setup( __file__, logdir=False, devices=[build_airtest_android_uri(config, serial=serial)], ) # 首先检查应用是否已安装 # 使用 ADBHelper（延迟初始化），ADBException 会自动向上传递 output = _get_adb_helper().shell(f"pm path {package_name}") package_paths = [ line.strip() for line in str(output or "").splitlines() if line.strip().startswith("package:") ] if not package_paths: logger.info(f"App {package_name} is not installed") return 'not_installed' # 应用已安装，尝试启动 start_app(package_name) wait_time = 10 # Wait for app launch for _ in range(wait_time): time.sleep(1) output = _get_adb_helper().shell(f"pidof {package_name}") if output.strip(): app_state = 'normal' break if app_state != 'normal': app_state = 'crash' logger.info(f"App state check result: {package_name} - {app_state}") time.sleep(2) stop_app(package_name) except Exception as e: logger.error(f"Error checking app state: {package_name} - {e}") import traceback logger.error(traceback.format_exc()) raise return app_state def download_app_chain(package_name, **kwargs): """ Smart download chain: iterates through available sources. Attempts to download and install. If install succeeds, treat it as success directly. Args: package_name (str): The package name of the app **kwargs: Additional arguments: - app_name (str): Name of the app - country (str): Country code for Google Play - local_path (str): Custom path for local import - target_account (str): Target Google account email for Google Play - available_sources (list): List of sources to try in order, e.g. ['google_play', 'local'] Returns: tuple: (success (bool), result (dict)) result format: { 'package_name': str, 'source': str, # 'local'/'google_play'/'pre-installed' 'state': str, # 'success'/'failed' 'run': str, # 'normal'/'crash'/'not_installed'/'unknown' 'error_code': int, # DownloadError 枚举值 'details': str, # 失败原因或成功说明 'crashed_source': str # 闪退的下载源（仅当 run='crash' 时） } """ available_sources = _normalize_available_sources(kwargs.get('available_sources', DEFAULT_SOURCES)) country_codes = normalize_country_codes(kwargs.get('country_codes') or kwargs.get('country')) platforms = available_sources errors = [] all_attempted_countries = [] logger.info(f"Checking if {package_name} is already installed...") run_state = check_app_state(package_name) if run_state == 'normal': return True, { 'package_name': package_name, 'source': 'pre-installed', 'state': 'success', 'run': 'normal', 'error_code': None, 'details': 'App already installed and runs normally' } for platform in platforms: logger.info(f"--- Attempting download from: {platform} ---") if platform == 'google_play': attempted_countries = [] country_errors = [] for country in country_codes: attempted_countries.append(country) all_attempted_countries = list(attempted_countries) attempt_kwargs = dict(kwargs) attempt_kwargs['country'] = country logger.info(f"--- Attempting Google Play country: {country} ---") success, message = download_app(platform, package_name, **attempt_kwargs) if not success: error_code = _parse_download_error(message, platform) country_errors.append({ 'source': _build_google_play_error_key(country), 'code': error_code, 'message': message, }) if error_code in GOOGLE_PLAY_COUNTRY_CONTINUE_ERRORS: continue break logger.info(f"Download/Install successful from {platform} ({country}). Skipping run-state verification.") local_path = kwargs.get('local_path') if local_path is None: local_path = str(_load_autool_config()["LOCAL_APK_PATH"]).strip() logger.info(f"Exporting APK to {local_path}") try: apks_export_import.export_multi_apk(package_name, local_path) except Exception as e: logger.error(f"Error exporting APK: {e}") return True, { 'package_name': package_name, 'source': platform, 'country': country, 'state': 'success', 'run': 'unknown', 'error_code': None, 'details': f'Installed from {platform} ({country})' } if country_errors and _can_fallback_to_local(country_errors): errors.extend(country_errors) if 'local' in platforms: continue return False, _build_failed_result( package_name, country_errors[-1], errors, attempted_countries=attempted_countries, ) if country_errors: errors.extend(country_errors) primary_error = next( ( item for item in reversed(country_errors) if item['code'] not in LOCAL_FALLBACK_ELIGIBLE_ERRORS ), country_errors[-1], ) return False, _build_failed_result( package_name, primary_error, errors, attempted_countries=attempted_countries, ) continue success, message = download_app(platform, package_name, **kwargs) if not success: error_code = _parse_download_error(message, platform) errors.append({ 'source': platform, 'code': error_code, 'message': message, }) continue logger.info(f"Download/Install successful from {platform}. Skipping run-state verification.") if platform != 'local': local_path = kwargs.get('local_path') if local_path is None: local_path = str(_load_autool_config()["LOCAL_APK_PATH"]).strip() logger.info(f"Exporting APK to {local_path}") try: apks_export_import.export_multi_apk(package_name, local_path) except Exception as e: logger.error(f"Error exporting APK: {e}") return True, { 'package_name': package_name, 'source': platform, 'state': 'success', 'run': 'unknown', 'error_code': None, 'details': f'Installed from {platform}' } logger.error("All download sources failed or app crashed on all versions.") if _is_country_locked_terminal(errors): return False, _build_failed_result( package_name, _pick_terminal_download_primary_error(errors), errors, attempted_countries=all_attempted_countries, ) last_error = errors[-1] if errors else { 'source': 'unknown', 'code': DownloadError.ALL_SOURCES_FAILED, 'message': 'All sources failed', 'run': 'unknown', } return False, { 'package_name': package_name, 'source': last_error['source'], 'state': 'failed', 'run': last_error.get('run', 'unknown'), 'error_code': last_error['code'], 'details': last_error['message'], 'errors': _build_errors_json(errors) } def download_app(platform, package_name, **kwargs): """ Unified interface for downloading apps. Args: platform (str): 'google_play' or 'local' package_name (str): The package name of the app (e.g., com.example.app) **kwargs: Additional arguments depending on platform: - app_name (str): Name of the app (useful for search in APKPure) - country (str): Country code (useful for Google Play) - local_path (str): Custom path for local import - target_account (str): Target Google account email for Google Play Returns: tuple: (success (bool), message (str)) """ logger.info(f"Requesting download: {package_name} from {platform}") try: downloader = _get_downloader(platform) if downloader: return downloader.start(package_name, **kwargs) else: msg = f"Unknown platform: {platform}" logger.error(msg) return False, msg except ADBException: raise except Exception as e: logger.error(f"Error during download process: {e}") return False, str(e) def stop_services(): """Stop all downloader services/connections""" logger.info("Stopping all downloader services...") for platform, downloader in _downloaders.items(): try: if hasattr(downloader, 'stop'): downloader.stop() except Exception as e: logger.error(f"Error stopping service for {platform}: {e}") # Optional: Clear the instances if you want to force re-initialization next time # _downloaders.clear() def reset_runtime_state(): """清空下载链缓存对象，避免模拟器重启后复用旧连接。""" global _adb_helper stop_services() _downloaders.clear() _adb_helper = None apks_export_import.reset_runtime_state() if __name__ == "__main__": # Example usage for testing download chain # This will try Google Play -> Local package_name = 'com.aakenya.testapp' # Optional args config = _load_autool_config() kwargs = { 'country': 'us', 'app_name': 'IKEA', 'local_path': str(config["LOCAL_APK_PATH"]).strip(), } logger.info(f"Starting download chain test for {package_name}") success, result = download_app_chain(package_name, **kwargs) print(f"\nFinal Chain Result: {success}") print(f"Details: {result}") stop_services()