# -*- encoding=utf8 -*- import os import sys import tempfile import zipfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) import logging from airtest.core.api import * logger = logging.getLogger(__name__) from config_loader import load_config as load_autool_config from adb_helper import ADBHelper from DroidBot.exceptions import ADBException def _load_autool_config(): return load_autool_config() class LocalFileImporter: def __init__(self): """初始化 LocalFileImporter，创建 ADB Helper 实例""" self.adb_helper = ADBHelper() def get_file_paths(self, folder): """Get all files in folder""" if not os.path.exists(folder): return [] return sorted( [ os.path.join(folder, f) for f in os.listdir(folder) if os.path.isfile(os.path.join(folder, f)) ] ) def _sort_apk_files(self, apk_files): return sorted( apk_files, key=lambda path: ( os.path.basename(path).lower() != 'base.apk', os.path.basename(path).lower(), ), ) def _collect_recursive_files(self, folder, suffix): matches = [] suffix_lower = suffix.lower() for root, _, files in os.walk(folder): for file_name in files: if file_name.lower().endswith(suffix_lower): matches.append(os.path.join(root, file_name)) return self._sort_apk_files(matches) if suffix_lower == '.apk' else sorted(matches) def _install_apk_files(self, package_name, apk_files): if len(apk_files) == 0: logger.info('No APK files found to install') return False, "no apk files" try: if len(apk_files) > 1: logger.info(f"Installing {len(apk_files)} APK files...") self.adb_helper.install_multiple(apk_files) else: logger.info("Installing single APK file...") self.adb_helper.install(apk_files[0]) logger.info(f"Import success: {package_name}") return True, "success" except ADBException as e: logger.error(f"ADB installation failed: {e}") if 'not connected' in str(e).lower(): raise return False, f"install failed: {e}" except Exception as e: logger.error(f"Error during import process: {e}") return False, str(e) def _push_obb_files(self, package_name, obb_files): if not obb_files: return remote_dir = f"/sdcard/Android/obb/{package_name}" logger.info(f"Preparing OBB directory: {remote_dir}") self.adb_helper.shell(f"mkdir -p {remote_dir}") for obb_path in obb_files: logger.info(f"Pushing OBB file: {obb_path}") self.adb_helper.run_cmd(['push', obb_path, remote_dir]) def _import_xapk(self, package_name, xapk_path): logger.info(f"Importing XAPK for {package_name}: {xapk_path}") try: with tempfile.TemporaryDirectory(prefix=f"{package_name.replace('.', '_')}_xapk_") as temp_dir: with zipfile.ZipFile(xapk_path, 'r') as archive: archive.extractall(temp_dir) apk_files = self._collect_recursive_files(temp_dir, '.apk') if not apk_files: logger.error("No APK files found inside XAPK") return False, "xapk missing apk files" obb_files = self._collect_recursive_files(temp_dir, '.obb') self._push_obb_files(package_name, obb_files) return self._install_apk_files(package_name, apk_files) except ADBException: raise except zipfile.BadZipFile: logger.error(f"Invalid XAPK archive: {xapk_path}") return False, "invalid xapk" except Exception as e: logger.error(f"Error during XAPK import process: {e}") return False, str(e) def start(self, package_name, local_path=None, local_apk_dir=None, local_apk_files=None, **kwargs): """ Main entry point for Local File Importer Installs APKs from a local directory (or network share) to the device. Args: package_name: Android package name local_path: Base directory for APK storage (defaults to config LOCAL_APK_PATH) local_apk_dir: Full path to the package-specific APK directory (from dispatcher) When set, used directly instead of os.path.join(local_path, package_name) local_apk_files: Optional list of expected files for validation """ # Determine source folder if local_apk_dir: folder = local_apk_dir logger.info(f"[LocalImport] 使用中控下发的APK路径: {folder}") else: if local_path is None: local_path = str(_load_autool_config()["LOCAL_APK_PATH"]).strip() folder = os.path.join(local_path, package_name) logger.info(f"[LocalImport] 使用默认路径: {folder}") logger.info(f"Attempting to import {package_name} from {folder}") logger.info(f"Folder Path: {folder}") if not os.path.isdir(folder): logger.error(f"Folder not found or invalid: {folder}") return False, "folder not found" if local_apk_files: expected_names = {str(f.get('filename') or '').strip() for f in local_apk_files} actual_names = {os.path.basename(p) for p in self.get_file_paths(folder)} missing = expected_names - actual_names if missing: logger.warning(f"Expected APK files missing from {folder}: {missing}") logger.info("Found files:") for path in self.get_file_paths(folder): logger.info(path) apk_files = self._sort_apk_files( [path for path in self.get_file_paths(folder) if path.lower().endswith('.apk')] ) has_base_apk = any(os.path.basename(path).lower() == 'base.apk' for path in apk_files) if has_base_apk: logger.info("Detected base.apk, using APK directory import") return self._install_apk_files(package_name, apk_files) xapk_files = [path for path in self.get_file_paths(folder) if path.lower().endswith('.xapk')] if not xapk_files: logger.error("No base.apk or XAPK files found in folder") return False, "base.apk not found and xapk missing" xapk_path = xapk_files[0] if len(xapk_files) > 1: logger.warning(f"Multiple XAPK files found, using first one: {xapk_path}") return self._import_xapk(package_name, xapk_path) def stop(self): """Stop any running processes (if any)""" # Local file import is usually blocking/synchronous in start(), # so there isn't a long-running background process to stop. # However, for consistency with other downloaders, we can log this. logger.info("LocalFileImporter: stop called (no active service to stop)") pass if __name__ == "__main__": package_name = "com.google.android.youtube" local_path = r"D:\DPI\autool\saveapk" importer = LocalFileImporter() try: success, message = importer.start(package_name, local_path) if success: print(f"Import completed successfully: {message}") else: print(f"Import failed: {message}") except Exception as e: print(f"Error: {e}")