# -*- encoding=utf8 -*- from __future__ import annotations from airtest.core.api import * from airtest.cli.parser import cli_setup from poco.drivers.android.uiautomation import AndroidUiautomationPoco import subprocess import os import re import csv import json import traceback import time import sys from dataclasses import dataclass from datetime import datetime # 导入统一的日志配置模块 from logging_config import setup_logging, get_logger from start_test import ( TaskRunner, TaskResult, load_config, EXIT_SUCCESS, EXIT_ERROR_GENERAL, EXIT_ERROR_USER, EXIT_ADB_ERROR, EXIT_NETWORK_ERROR, EXIT_SIMULATOR_ERROR, EXIT_EXPLORATION_STUCK, EXIT_APP_CRASH_ERROR, EXIT_APP_NEED_UPDATE, EXIT_APP_LAUNCH_ERROR, STUCK_REASON_TO_ERROR_TYPE, ) from result_codes import ErrorInfo, STUCK_TO_ERROR, ErrorCategory, InfraError, AppError, BusinessError, DownloadError from utils_android.Manager.mumu_manager import MuMuManager from utils_android.Manager.physical_device_manager import PhysicalDeviceManager from utils_android.Manager.pcapdroid_manager import PCAPManager from utils_android.Manager.data_manager import DataManager from utils_android.Manager.network_manager import NetworkWatchdog, NetworkException, check_network, wait_for_network from utils_android.device_config import ( build_airtest_android_uri, ensure_android_wireless_connected, get_android_device_serial, ) from utils_android.download_app.download_controller import ( download_app_chain, stop_services, reset_runtime_state as reset_download_runtime_state, ) from utils_android.download_app.country_codes import normalize_country_codes from DroidBot.exceptions import ADBException # ========================================== # 0. NullTaskWorker（单应用模式，不连 Redis） # ========================================== class NullTaskWorker: """不连接 Redis 的 Worker 桩，用于单应用本地测试模式。""" def init(self): return None def report(self, report_data): status = report_data.get("status", "?") task_key = report_data.get("task_key", "?") error_info = report_data.get("error", "") if error_info: print(f"[NullWorker] Report task={task_key} status={status} error={error_info}") else: print(f"[NullWorker] Report task={task_key} status={status}") return None def retry(self): return True def event(self, event_data): pass # ========================================== # 1. 日志重定向类 (捕获所有 stdout/stderr) # ========================================== class TeeOutput: """同时将输出写入文件和控制台""" def __init__(self, filepath, original_stream): self.file = open(filepath, 'a', encoding='utf-8') self.original_stream = original_stream def write(self, data): self.file.write(data) self.file.flush() self.original_stream.write(data) self.original_stream.flush() def flush(self): self.file.flush() self.original_stream.flush() def close(self): self.file.close() def isatty(self): return False # 配置日志目录和文件路径 log_dir = os.path.join(os.path.dirname(__file__), 'logs') os.makedirs(log_dir, exist_ok=True) log_filename = datetime.now().strftime("batch_run_%Y%m%d_%H%M%S.log") log_filepath = os.path.join(log_dir, log_filename) # 重定向 stdout 和 stderr 到日志文件 sys.stdout = TeeOutput(log_filepath, sys.stdout) sys.stderr = TeeOutput(log_filepath, sys.stderr) # 使用统一的日志配置 setup_logging( level='INFO', log_file=log_filepath, debug_mode=False ) logger = get_logger(__name__) logger.info(f"日志文件路径: {log_filepath}") # ========================================== # 2. 统计管理 (BatchStatistics) # ========================================== class BatchStatistics: """批量测试统计管理器""" def __init__(self): self.start_time = datetime.now() self.records = [] self.current_round = 0 def add_record(self, round_num: int, app_name: str, package_name: str, result: TaskResult, is_retry: bool = False): """添加一条测试记录""" self.records.append({ "round": round_num, "app_name": app_name, "package_name": package_name, "status": result.status, "exit_code": result.exit_code, "error_reason": result.error_reason, "droidbot_steps": result.droidbot_steps, "guiagent_steps": result.guiagent_steps, "total_steps": result.total_steps, "duration_seconds": round(result.duration_seconds, 2), "is_retry": is_retry, "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) def get_current_statistics(self, result: TaskResult, round_num: int, is_retry: bool = False, download_source: str = None) -> dict: """获取当前任务的统计数据，用于上报""" return { "round": round_num, "exit_code": result.exit_code, "error_reason": result.error_reason, "droidbot_steps": result.droidbot_steps, "guiagent_steps": result.guiagent_steps, "total_steps": result.total_steps, "duration_seconds": round(result.duration_seconds, 2), "is_retry": is_retry, "download_source": download_source, "num_nodes": result.num_nodes, "num_reached_activities": result.num_reached_activities, "app_num_total_activities": result.app_num_total_activities } def get_round_summary(self, round_num: int) -> dict: """获取某轮的统计摘要""" round_records = [r for r in self.records if r["round"] == round_num] success_count = sum(1 for r in round_records if r["status"] == "SUCCESS") failed_count = sum(1 for r in round_records if r["status"] == "FAILED") return { "total": len(round_records), "success": success_count, "failed": failed_count, "retry_count": sum(1 for r in round_records if r["is_retry"]) } @dataclass class TaskContext: task: dict app_name: str package_name: str task_key: str country_codes: list available_sources: list local_apk_dir: str = None local_apk_files: list = None download_source: str = None output_dir: str = None final_result: TaskResult = None status_to_report: str = "failed" error_info: object = None collect_retry_count: int = 0 keep_app_installed: bool = False enable_app_block: bool = False app_block_mode: str = "set" app_block_tag: str = None class FinalizeRecoveryFailed(RuntimeError): """finalize 阶段环境恢复失败后终止当前轮。""" # ========================================== # 3. 执行部分 (TestExecutor) # ========================================== class TestExecutor: """自动化测试的执行引擎""" BROWSER_PACKAGE_CANDIDATES = ( "com.android.chrome", "com.android.chromium", "com.android.browser", ) def __init__(self, stats: BatchStatistics = None, worker=None): self.config = load_config() self.data = DataManager(self.config) self.device = MuMuManager(self.config) if self.config.get('IS_EMULATOR', True) else PhysicalDeviceManager(self.config) self.mumu = self.device self.network_watchdog = NetworkWatchdog() self._adb_helper = None self._pcap = None self.task_runner = TaskRunner(self.config) self.stats = stats or BatchStatistics() self.current_round = 0 self._base_packages = None self.device_serial = get_android_device_serial(self.config) self.app_block_app_list = self.config.get('APP_BLOCK_APP_LIST') self.app_block_url_lib = self.config.get('APP_BLOCK_URL_LIB') self._worker = worker @property def adb_helper(self): """延迟初始化 ADB Helper""" if self._adb_helper is None: from utils_android.download_app.adb_helper import ADBHelper self._adb_helper = ADBHelper(self.device_serial) return self._adb_helper @property def pcap(self): """延迟初始化 PCAPManager（依赖 adb_helper）""" if self._pcap is None: self._pcap = PCAPManager(self.adb_helper, root_capture=self.device.device_type == 'emulator') return self._pcap @property def device_label(self) -> str: return "模拟器" if self.device.device_type == "emulator" else "真机" @property def is_emulator_device(self) -> bool: return self.device.device_type == "emulator" def _reset_runtime_state(self): """设备重启后丢弃所有持有旧设备连接的缓存对象。""" self._adb_helper = None self._pcap = None reset_download_runtime_state() def _reinitialize_after_restart(self): self._reset_runtime_state() print(f"[INFO] {self.device_label}已恢复，重新初始化 Airtest 连接...") try: return self.setup() except ADBException as e: logger.error(f"重启后重新初始化 Airtest 失败: {e}") return False except Exception as e: logger.error(f"重启后初始化运行时失败: {e}") return False def _restart_after_adb_fault(self): print(f"[CRITICAL] 检测到 ADB 断联，正在重启{self.device_label}...") logger.error(f"检测到 ADB 断联，触发{self.device_label}重启恢复流程") if not self.device.restart(): self._reset_runtime_state() return False return self._reinitialize_after_restart() def _load_base_packages(self): """加载纯净镜像的包名白名单（仅加载一次）""" if self._base_packages is not None: return self._base_packages if not self.is_emulator_device: logger.info("当前为真机模式，跳过纯净镜像包名白名单校验") self._base_packages = set() return self._base_packages packages_file = os.path.join(os.path.dirname(__file__), 'doc', 'android_packages.txt') if not os.path.exists(packages_file): logger.warning(f"纯净镜像包名文件不存在: {packages_file}") self._base_packages = set() return self._base_packages packages = set() with open(packages_file, 'r', encoding='utf-8') as f: for line in f: line = line.strip() if line.startswith('package:'): packages.add(line.split('package:')[1].strip()) elif line: # 兼容没有 package: 前缀的格式 packages.add(line) logger.info(f"已加载纯净镜像包名白名单: {len(packages)} 个包") self._base_packages = packages return self._base_packages def _get_third_party_packages(self): """获取设备上已安装的第三方包列表（通过 adb_helper 封装调用）""" output = self.adb_helper.shell("pm list package") return [line.split("package:")[1].strip() for line in output.splitlines() if "package:" in line] def _stop_browser_apps(self): """停止所有已知浏览器候选包，兼容模拟器与真机。""" for package_name in self.BROWSER_PACKAGE_CANDIDATES: try: stop_app(package_name) except Exception as exc: logger.debug(f"停止浏览器 {package_name} 失败，忽略: {exc}") def _resolve_pocoservice_apk(self) -> str: """定位 Poco 随包提供的 Android 服务 APK。""" import poco poco_dir = os.path.dirname(os.path.abspath(poco.__file__)) return os.path.join(poco_dir, "drivers", "android", "lib", "pocoservice-debug.apk") def uninstall_extra_apps(self, current_package: str): """卸载白名单和当前任务包名之外的多余应用""" base_packages = self._load_base_packages() if not base_packages: logger.warning("白名单为空，跳过卸载多余应用") return # 获取当前设备上所有已安装的包 installed = self._get_third_party_packages() # 计算需要卸载的包：不在白名单中，且不是当前任务的包 to_uninstall = [ pkg for pkg in installed if pkg not in base_packages and pkg != current_package ] if not to_uninstall: logger.info("没有需要卸载的多余应用") return logger.info(f"检测到 {len(to_uninstall)} 个多余应用，开始卸载: {to_uninstall}") for pkg in to_uninstall: if self.adb_helper.uninstall(pkg): logger.info(f"已卸载: {pkg}") continue logger.warning(f"卸载 {pkg} 失败") def setup(self): if not cli_setup(): try: ensure_android_wireless_connected(self.config, self.device_serial) auto_setup( __file__, logdir=False, devices=[build_airtest_android_uri(self.config, serial=self.device_serial)], ) except Exception as e: print(f"设备连接失败: {e}") return False if "com.netease.open.pocoservice" not in self._get_third_party_packages(): install(self._resolve_pocoservice_apk()) sleep(5) start_app("com.netease.open.pocoservice") sleep(2) return True def _create_task_context(self, task: dict) -> TaskContext: country_codes = normalize_country_codes(task.get('country_codes') or task.get('country_code', '')) return TaskContext( task=task, app_name=task.get('app_name', 'Unknown App'), package_name=task.get('package_name', ''), task_key=task.get('task_key', ''), country_codes=country_codes, available_sources=task.get('available_sources', ['google_play', 'local']), local_apk_dir=task.get('local_apk_dir', '') or None, local_apk_files=task.get('local_apk_files', []) or None, error_info=ErrorInfo.success(), keep_app_installed=bool(task.get('keep_app_installed', False)), enable_app_block=bool(task.get('blocked', False)), app_block_mode=task.get('app_block_mode', 'set'), app_block_tag=task.get('app_block_tag') or None, ) def _error_type_str(self, error) -> str: if hasattr(error, 'category') and hasattr(error, 'code'): return f"{error.category.name}/{error.code}" if isinstance(error, dict) and error.get('category') is not None and error.get('code') is not None: return f"{error['category']}/{error['code']}" return "UNEXPECTED/0" def _report_terminal(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, *, status: str, failed_stage: str, error, error_message: str = None, download_errors: dict = None, extra_error: dict = None, **extra): error_payload = error.to_report_dict() if hasattr(error, 'to_report_dict') else dict(error or {}) if extra_error: error_payload.update(extra_error) if error_message is None: error_message = error_payload.get('reason', '') mon.finish_task( status, failed_stage=failed_stage, error_type=self._error_type_str(error), error_message=error_message, download_errors=download_errors, ) return worker.report( mon.build_report_payload( status, error_payload, download_errors=download_errors, **extra, ) ) def _prepare_download(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker): mon.start_stage("download", "running_download") while True: device_ok, was_restarted = self.device.ensure_running() if not device_ok: print(f"[ERROR] 跳过 {ctx.app_name}: {self.device_label}不可用") mon.finish_stage("download", status="failed") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="download", error=ErrorInfo.infra( InfraError.EMULATOR_START_FAILED if self.is_emulator_device else InfraError.ADB_ERROR ), error_message=f"{self.device_label} unavailable", ) try: if was_restarted: mon.mark("device_restarted") setup_ok = self._reinitialize_after_restart() if not setup_ok: print(f"[ERROR] 跳过 {ctx.app_name}: Airtest 重新连接失败") mon.finish_stage("download", status="failed") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="download", error=ErrorInfo.infra(InfraError.AIRTEST_INIT_FAILED), error_message="airtest reinit failed", ) mon.mark("airtest_reinitialized") worker.retry() self.uninstall_extra_apps(ctx.package_name) return None except ADBException as e: logger.error(f"准备下载阶段检测到 ADB 异常: {e}") print(f"[ERROR] 清理多余应用时 ADB 断联 {ctx.app_name}，尝试恢复...") if not self._restart_after_adb_fault(): mon.finish_stage("download", status="failed") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="download", error=ErrorInfo.infra( InfraError.EMULATOR_RECOVERY_FAILED if self.is_emulator_device else InfraError.ADB_ERROR, str(e), ), error_message=f"{self.device_label} recovery failed during cleanup", ) mon.mark("download_recovered") worker.retry() def _build_download_kwargs(self, ctx: TaskContext) -> dict: kwargs = { 'app_name': ctx.app_name, 'country': ctx.country_codes[0], 'country_codes': ctx.country_codes, 'local_path': self.config.get('LOCAL_APK_PATH'), 'apkpure_path': self.config.get('APKPURE_APK_PATH'), 'target_account': '', 'available_sources': ctx.available_sources, } if ctx.local_apk_dir: print(f"[Task] 收到本地APK路径: {ctx.local_apk_dir}") print(f"[Task] 本地APK文件: {ctx.local_apk_files}") kwargs['local_apk_dir'] = ctx.local_apk_dir else: print(f"[Task] 未收到本地APK路径 (available_sources={ctx.available_sources})") if ctx.local_apk_files: kwargs['local_apk_files'] = ctx.local_apk_files return kwargs def _handle_download_result(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, dl_success: bool, dl_result: dict): if not dl_success: error_code = dl_result.get('error_code', DownloadError.OTHER) errors_json = dl_result.get('errors', {}) attempted_countries = dl_result.get('attempted_countries', []) mon.finish_stage("download", status="failed", download_source=dl_result.get('source', 'unknown')) print(f"[ERROR] 应用下载失败 {ctx.app_name}: {errors_json}") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="download", error=ErrorInfo.download(error_code), error_message=json.dumps(errors_json, ensure_ascii=False), download_errors=errors_json, attempted_countries=attempted_countries, ) if dl_result.get('run') == 'crash': details = dl_result.get('details', 'App crashed') crashed_source = dl_result.get('crashed_source', 'unknown') mon.finish_stage("download", status="failed", download_source=dl_result.get('source', 'unknown')) print(f"[ERROR] 应用安装成功但无法运行 {ctx.app_name}: {details}") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="download", error=ErrorInfo.app(AppError.CRASH, details), error_message=details, extra_error={"crashed_source": crashed_source}, ) print(f"[INFO] 应用准备就绪 (来源: {dl_result.get('source')}, 状态: {dl_result.get('state')})") ctx.download_source = dl_result.get('source', 'unknown') mon.finish_stage("download", status="success", download_source=ctx.download_source) mon.start_stage("collect", "running_collect") return None def _recover_download(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, dl_kwargs: dict): recovery_success = self.device.recover_from_frozen() if self.device.is_player_frozen() else self.device.restart() if not recovery_success: self._reset_runtime_state() mon.finish_stage("download", status="failed") print(f"[ERROR] {self.device_label}恢复失败，无法继续测试: {ctx.app_name}") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="download", error=ErrorInfo.infra( InfraError.EMULATOR_RECOVERY_FAILED if self.is_emulator_device else InfraError.ADB_ERROR ), error_message=f"{self.device_label} recovery failed during download", ) mon.mark("download_recovered") logger.info(f"{self.device_label}恢复成功，重新初始化连接...") if not self._reinitialize_after_restart(): mon.finish_stage("download", status="failed") print(f"[ERROR] 跳过 {ctx.app_name}: Airtest 重新连接失败") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="download", error=ErrorInfo.infra(InfraError.AIRTEST_INIT_FAILED), error_message="airtest reinit failed after download recovery", ) worker.retry() logger.info("重新尝试下载应用...") try: dl_success, dl_result = download_app_chain(ctx.package_name, **dl_kwargs) return self._handle_download_result(ctx, mon, worker, dl_success, dl_result) except Exception as e: mon.finish_stage("download", status="failed") print(f"[ERROR] 重试下载时出现异常 {ctx.app_name}: {e}") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="download", error=ErrorInfo.download(DownloadError.OTHER, str(e)), error_message=str(e), ) def _execute_download(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker): print(f"[INFO] 正在启动智能下载链条: {ctx.app_name} ({ctx.package_name})") dl_kwargs = self._build_download_kwargs(ctx) try: dl_success, dl_result = download_app_chain(ctx.package_name, **dl_kwargs) return self._handle_download_result(ctx, mon, worker, dl_success, dl_result) except (ADBException, IndexError) as e: logger.error(f"下载过程中检测到 ADB 异常: {e}") print(f"[ERROR] 下载应用时 ADB 断联 {ctx.app_name}，尝试恢复...") return self._recover_download(ctx, mon, worker, dl_kwargs) except Exception as e: traceback.print_exc() mon.finish_stage("download", status="failed") print(f"[ERROR] 下载过程出现异常 {ctx.app_name}: {e}") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="download", error=ErrorInfo.download(DownloadError.OTHER, str(e)), error_message=str(e), ) def _prepare_collect(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker): print("[INFO] 清理下载相关进程 (Google Play, Browser, APKPure, PocoService)...") stop_services() stop_app("com.android.vending") self._stop_browser_apps() stop_app("com.apkpure.aegon") stop_app("com.netease.open.pocoservice") timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") ctx.output_dir = os.path.join(self.config['OUTPUT_BASE_DIR'], f"{ctx.package_name}_{timestamp}") # 根据当前任务上下文重新初始化 PCAPManager（应用阻塞时关闭 root 抓包） root_capture = self.device.device_type == 'emulator' and not ctx.enable_app_block app_blocker = None if ctx.enable_app_block and self.app_block_app_list and self.app_block_url_lib: from utils_android.Manager.pcapdroid_manager import AppBlocker app_blocker = AppBlocker( self.adb_helper, app_list_file=self.app_block_app_list, url_lib_file=self.app_block_url_lib, ) self._pcap = PCAPManager( self.adb_helper, root_capture=root_capture, app_blocker=app_blocker, ) # 导入应用阻塞规则 if ctx.enable_app_block: print(f"[INFO] 正在导入应用阻塞规则 (package={ctx.package_name}, tag={ctx.app_block_tag}, mode={ctx.app_block_mode})") block_ok = self._pcap.import_app_blocklist( ctx.package_name, mode=ctx.app_block_mode, tag=ctx.app_block_tag, ) if block_ok: print(f"[INFO] 应用阻塞规则导入成功") else: print(f"[WARNING] 应用阻塞规则导入失败或该应用无规则") if self.pcap.start_capture(ctx.package_name): mon.mark("pcap_started") return None mon.finish_stage("collect", status="failed") print(f"[ERROR] 跳过 {ctx.app_name}: PCAP 启动失败") return self._report_terminal( ctx, mon, worker, status="failed", failed_stage="collect", error=ErrorInfo.infra(InfraError.PCAP_START_FAILED), error_message="pcap start failed", ) def _pcap_sync_wrapper(self, pkg_name): print(f"[Callback] 检测到 PCAPDroid 停止，正在推送流量文件: {pkg_name}") self.data.sync_latest_traffic_file(pkg_name) def _run_collect_task(self, ctx: TaskContext, mon: TaskMonitor) -> TaskResult: try: result = self.task_runner.run_task( ctx.package_name, ctx.app_name, ctx.output_dir, pcap_callback=self._pcap_sync_wrapper, enable_app_block=ctx.enable_app_block, ) except KeyboardInterrupt as e: if e.args and isinstance(e.args[0], TaskResult): result = e.args[0] self.stats.add_record(self.current_round, ctx.app_name, ctx.package_name, result) raise return result def _retry_collect(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, recover_fn, failure_error, retry_error_builder): if not recover_fn(): self._reset_runtime_state() ctx.status_to_report = 'failed' ctx.error_info = failure_error return mon.mark("collect_recovered") if not self._reinitialize_after_restart(): ctx.status_to_report = 'failed' ctx.error_info = ErrorInfo.infra(InfraError.AIRTEST_INIT_FAILED) return worker.retry() retry_result = self.task_runner.run_task( ctx.package_name, ctx.app_name, ctx.output_dir, pcap_callback=self._pcap_sync_wrapper, enable_app_block=ctx.enable_app_block, ) self.stats.add_record(self.current_round, ctx.app_name, ctx.package_name, retry_result, is_retry=True) ctx.final_result = retry_result ctx.status_to_report = 'success' if retry_result.status == 'SUCCESS' else 'failed' ctx.collect_retry_count = 1 if retry_result.status != "SUCCESS": ctx.error_info = retry_error_builder(retry_result) def _resolve_collect_outcome(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker): result = ctx.final_result ctx.status_to_report = 'success' if result.status == 'SUCCESS' else 'failed' ctx.error_info = ErrorInfo.success() ctx.collect_retry_count = 0 if result.exit_code == EXIT_SIMULATOR_ERROR: print(f"[CRITICAL] 检测到{self.device_label}异常 (code {result.exit_code})，触发紧急恢复流程...") self._retry_collect( ctx, mon, worker, self.device.recover if self.is_emulator_device else self.device.restart, ErrorInfo.infra( InfraError.EMULATOR_RECOVERY_FAILED if self.is_emulator_device else InfraError.ADB_ERROR ), lambda retry_result: ErrorInfo.infra( InfraError.EMULATOR_CRASH if self.is_emulator_device else InfraError.ADB_ERROR, retry_result.error_reason, ), ) elif result.exit_code == EXIT_ADB_ERROR: print(f"[CRITICAL] 检测到 ADB 断联 (code {result.exit_code})，正在重启{self.device_label}...") self._retry_collect( ctx, mon, worker, self.device.restart, ErrorInfo.infra( InfraError.EMULATOR_RECOVERY_FAILED if self.is_emulator_device else InfraError.ADB_ERROR ), lambda retry_result: ErrorInfo.infra(InfraError.ADB_ERROR, retry_result.error_reason), ) elif result.exit_code == EXIT_NETWORK_ERROR: print(f"[ERROR] 网络异常，跳过当前应用: {ctx.app_name} ({result.error_reason})") ctx.status_to_report = 'failed' ctx.error_info = ErrorInfo.infra(InfraError.NETWORK_ERROR, result.error_reason) elif result.exit_code == EXIT_APP_CRASH_ERROR: if ctx.enable_app_block: print(f"[WARN] Block任务应用闪退，视为正常结束: {ctx.app_name} ({result.error_reason})") ctx.status_to_report = 'success' ctx.error_info = ErrorInfo.success() else: print(f"[ERROR] 应用闪退，跳过当前应用: {ctx.app_name} ({result.error_reason})") ctx.status_to_report = 'failed' ctx.error_info = ErrorInfo.app(AppError.CRASH, result.error_reason) elif result.exit_code == EXIT_APP_NEED_UPDATE: print(f"[ERROR] 应用需更新，跳过当前应用: {ctx.app_name} ({result.error_reason})") ctx.status_to_report = 'failed' ctx.error_info = ErrorInfo.app(AppError.NEED_UPDATE, result.error_reason) elif result.exit_code == EXIT_APP_LAUNCH_ERROR: print(f"[ERROR] 启动异常，跳过当前应用: {ctx.app_name} ({result.error_reason})") ctx.status_to_report = 'failed' ctx.error_info = ErrorInfo.app(AppError.LAUNCH_ERROR, result.error_reason) elif result.exit_code == EXIT_EXPLORATION_STUCK: print(f"[ERROR] 探索停滞，跳过当前应用: {ctx.app_name} ({result.error_reason})") ctx.status_to_report = 'failed' ctx.error_info = ErrorInfo.from_stuck_reason(result.stuck_reason_code, ctx.final_result.guiagent_message or result.error_reason) elif result.status != "SUCCESS": print(f"[ERROR] 任务执行失败: {ctx.app_name} ({result.error_reason})") ctx.status_to_report = 'failed' ctx.error_info = ErrorInfo(ErrorCategory.BUSINESS_ERROR, BusinessError.OTHER, result.error_reason) if ctx.error_info.category != ErrorCategory.SUCCESS: return if ctx.final_result.guiagent_message: has_login_failed = "登录失败" in ctx.final_result.guiagent_message has_register_failed = "注册失败" in ctx.final_result.guiagent_message if has_login_failed and has_register_failed: ctx.error_info = ErrorInfo.business(BusinessError.LOGIN_FAILED) elif has_login_failed: ctx.error_info = ErrorInfo.business(BusinessError.LOGIN_FAILED) elif has_register_failed: ctx.error_info = ErrorInfo.business(BusinessError.REGISTER_FAILED) if ctx.final_result.login_count == 0 and ctx.final_result.register_count == 0: ctx.error_info = ErrorInfo.business(BusinessError.NO_SCENARIO) def _report_task_result(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker): mon.finish_task( ctx.status_to_report, failed_stage=None if ctx.status_to_report == 'success' else 'collect', error_type=None if ctx.status_to_report == 'success' else self._error_type_str(ctx.error_info), error_message=None if ctx.status_to_report == 'success' else (ctx.error_info.reason or ctx.final_result.error_reason), retry_count=ctx.collect_retry_count, ) report_data = mon.build_report_payload( ctx.status_to_report, ctx.error_info if ctx.status_to_report != 'success' else None, metrics={ "login_count": ctx.final_result.login_count, "register_count": ctx.final_result.register_count, "guiagent_message": ctx.final_result.guiagent_message, "scenario_triggered": ctx.final_result.login_count > 0 or ctx.final_result.register_count > 0, "stuck_reason_code": ctx.final_result.stuck_reason_code, }, statistics=self.stats.get_current_statistics( ctx.final_result, self.current_round, download_source=ctx.download_source, ), ) print(f"任务完成，状态: {ctx.status_to_report}, 错误: {ctx.error_info.category.name}/{ctx.error_info.code}") return worker.report(report_data) def _run_one_task(self, worker: TaskWorker, ctx: TaskContext): print(f"\n>>>> 开始处理任务: {ctx.app_name} <<<<") print(f" 包名: {ctx.package_name}") print(f" 任务键: {ctx.task_key}") print(f" 国家码: {ctx.country_codes}") print(f" 可用下载源: {ctx.available_sources}") print(f" 应用阻塞: {ctx.enable_app_block} (mode={ctx.app_block_mode}, tag={ctx.app_block_tag})") # 每个任务开始前：清除上一个任务遗留的 blocklist 并重置 PCAPManager if self._pcap is not None: try: self._pcap.clear_app_blocklist() except Exception as e: logger.warning(f"清除上一个任务的 blocklist 失败: {e}") self._pcap = None from utils_android.TaskWorker.task_monitor import TaskMonitor # 延迟导入，避免单应用模式依赖 Redis mon = TaskMonitor(worker, ctx.task_key) mon.start_task() self.network_watchdog.check_and_raise() next_task = self._prepare_download(ctx, mon, worker) if next_task is not None: return next_task, mon next_task = self._execute_download(ctx, mon, worker) if next_task is not None: return next_task, mon next_task = self._prepare_collect(ctx, mon, worker) if next_task is not None: return next_task, mon ctx.final_result = self._run_collect_task(ctx, mon) self.stats.add_record(self.current_round, ctx.app_name, ctx.package_name, ctx.final_result) self._resolve_collect_outcome(ctx, mon, worker) mon.finish_stage( "collect", status="success" if ctx.status_to_report == 'success' else "failed", duration_seconds=round(ctx.final_result.duration_seconds, 2), ) stop_app(ctx.package_name) start_app("com.netease.open.pocoservice") self.pcap.stop_capture() sleep(2) return self._report_task_result(ctx, mon, worker), mon def _stop_current_task(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, error): from utils_android.TaskWorker.task_monitor import TaskMonitor # noqa: F811 failed_stage = None if mon and "collect" in mon.stage_started_at: failed_stage = "collect" mon.finish_stage("collect", status="failed") elif mon and "download" in mon.stage_started_at: failed_stage = "download" mon.finish_stage("download", status="failed") return self._report_terminal( ctx, mon or TaskMonitor(worker, ctx.task_key), worker, status="stop", failed_stage=failed_stage, error=error, error_message=error.get('reason') if isinstance(error, dict) else error.reason, ) def _finalize_task(self, ctx: TaskContext, worker: TaskWorker): traffic_synced = False log_synced = False if ctx.package_name: if ctx.keep_app_installed: print(f"[INFO] 保留应用: {ctx.package_name}") else: print(f"[INFO] 卸载应用: {ctx.package_name}") if self.adb_helper.uninstall(ctx.package_name): print(f"[INFO] 应用 {ctx.package_name} 卸载成功") else: print(f"[WARNING] 卸载应用 {ctx.package_name} 失败") traffic_synced = bool(self.data.sync_latest_traffic_file(ctx.package_name)) if log_filepath: log_synced = bool(self.data.sync_log_file(log_filepath)) # 清除应用阻塞规则 if ctx.enable_app_block: try: print("[INFO] 正在清除应用阻塞规则...") if self._pcap is not None: self._pcap.clear_app_blocklist() else: from utils_android.Manager.pcapdroid_manager import AppBlocker AppBlocker(self.adb_helper).clear_blocklist() except Exception as e: logger.warning(f"清除应用阻塞规则失败: {e}") if not ctx.output_dir: worker.event({ "event_type": "artifacts_synced", "task_key": ctx.task_key, "package_name": ctx.package_name, "traffic_synced": traffic_synced, "log_synced": log_synced, }) return worker.event({ "event_type": "artifacts_synced", "task_key": ctx.task_key, "package_name": ctx.package_name, "traffic_synced": traffic_synced, "log_synced": log_synced, }) def _finalize_with_recovery(self, ctx: TaskContext, worker: TaskWorker, next_task): try: self._finalize_task(ctx, worker) return next_task except ADBException as e: logger.error(f"finalize 阶段检测到 ADB 异常: {e}") print(f"[CRITICAL] finalize 卸载阶段 ADB 断联 {ctx.app_name}，尝试重启{self.device_label}恢复...") if self._restart_after_adb_fault(): print("[INFO] finalize 阶段环境恢复成功，继续执行后续任务") return next_task if next_task: logger.warning("finalize 恢复失败，重新计时已领取的下一任务") worker.retry() raise FinalizeRecoveryFailed(f"finalize recovery failed: {e}") def run(self, initial_task=None): from utils_android.TaskWorker import TaskWorker # 延迟导入，避免单应用模式依赖 Redis 配置 worker = self._worker if self._worker is not None else TaskWorker() logger.info("启动后台网络监控...") self.network_watchdog.start() print("\n[1] 初始化Worker...") if initial_task is not None: task = initial_task print(f"[INFO] 使用注入的初始任务: {task.get('package_name', '?')}") else: task = worker.init() if not task: print("初始化失败，程序退出") self.network_watchdog.stop() return print("\n[2] 开始处理任务...") print("按 Ctrl+C 停止Worker\n") fatal_error = None interrupted = False while task: if task.get('package_name', '') == '-1': logger.error(f"任务包名异常, 任务键: {task.get('task_key', '')}") break ctx = self._create_task_context(task) if task.get('traffic_root'): self.config['TRAFFIC_DATA_SHARE'] = task['traffic_root'] mon = None next_task = None interrupted_this_task = False try: next_task, mon = self._run_one_task(worker, ctx) except NetworkException as e: logger.error(f"后台监控检测到网络异常: {e}") lan_info = f"{e.ip_address}局域网" if e.ip_address else "未知局域网" mac_info = f"{e.mac_address}mac" if e.mac_address else "未知mac" stop_error = ErrorInfo.infra(InfraError.NETWORK_ERROR, f"{lan_info}，{mac_info} 网络错误") next_task = self._stop_current_task(ctx, mon, worker, stop_error) logger.info("已停止当前任务，等待网络恢复...") self.network_watchdog.stop() except KeyboardInterrupt: logger.info("\n收到中断信号,程序退出。") next_task = self._stop_current_task(ctx, mon, worker, {"category": "USER_INTERRUPT", "code": 0, "reason": "User interrupted"}) interrupted_this_task = True except Exception as e: print(f"处理任务时出错: {e}") traceback.print_exc() failed_stage = None if mon and "collect" in mon.stage_started_at: failed_stage = "collect" mon.finish_stage("collect", status="failed") elif mon and "download" in mon.stage_started_at: failed_stage = "download" mon.finish_stage("download", status="failed") next_task = self._report_terminal( ctx, mon or TaskMonitor(worker, ctx.task_key), worker, status="failed", failed_stage=failed_stage, error={"category": "UNEXPECTED", "code": 0, "reason": str(e)}, error_message=str(e), ) try: task = self._finalize_with_recovery(ctx, worker, next_task) except FinalizeRecoveryFailed as e: fatal_error = e break if interrupted_this_task: interrupted = True break logger.info("任务循环结束，停止后台网络监控...") self.network_watchdog.stop() if fatal_error: raise fatal_error if interrupted: raise KeyboardInterrupt print("\n所有任务已完成或无更多任务") if __name__ == '__main__': import argparse parser = argparse.ArgumentParser( description="autool 批量/单应用测试运行器", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="""示例: # 标准批量模式（连接 Redis 中控端） python batch_run.py # 单应用本地测试模式 python batch_run.py --single --package com.google.android.youtube python batch_run.py --single --package com.tencent.mm --app-name "微信" --country CN --keep-app """, ) parser.add_argument( "--single", action="store_true", help="启用单应用本地测试模式（不连接 Redis 中控端）", ) parser.add_argument( "--package", type=str, default=None, help="应用包名（--single 模式下必需）", ) parser.add_argument( "--app-name", type=str, default=None, help="应用显示名称（可选，默认取包名最后一段）", ) parser.add_argument( "--country", type=str, default="US", help="国家码，默认 US", ) parser.add_argument( "--keep-app", action="store_true", help="测试完成后保留应用（默认卸载）", ) args = parser.parse_args() # 清理自定义参数，避免 airtest 的 cli_setup 解析时报 unrecognized arguments sys.argv = [sys.argv[0]] if args.single: if not args.package: parser.error("--single 模式必须指定 --package") app_name = args.app_name or args.package.split(".")[-1] task_key = f"single-{args.package}-{datetime.now().strftime('%Y%m%d%H%M%S')}" task = { "app_name": app_name, "package_name": args.package, "task_key": task_key, "country_codes": [args.country.upper()], "available_sources": ["google_play", "local"], "keep_app_installed": args.keep_app, } config = load_config() print(f"\n{'='*50}") print(f"[单应用模式]") print(f" 应用: {app_name}") print(f" 包名: {args.package}") print(f" 国家码: {args.country.upper()}") print(f" 保留应用: {args.keep_app}") print(f"{'='*50}\n") if not check_network(): logger.warning("初始网络检查失败，等待网络恢复...") wait_for_network(check_interval=5) executor = TestExecutor(worker=NullTaskWorker()) executor.setup() executor.run(initial_task=task) print("\n[单应用模式] 测试完成。") else: round_count = 0 stats = None config = load_config() while True: try: if not check_network(): logger.warning("初始网络检查失败，等待网络恢复...") wait_for_network(check_interval=5) if stats is None: stats = BatchStatistics() executor = TestExecutor(stats=stats) executor.setup() round_count += 1 executor.current_round = round_count logger.info(f"========== 开始第 {round_count} 轮测试 ==========") executor.run() summary = stats.get_round_summary(round_count) logger.info(f"========== 第 {round_count} 轮测试完成 ==========") logger.info(f"本轮统计: 总计 {summary['total']} 个应用, 成功 {summary['success']}, 失败 {summary['failed']}, 重试 {summary['retry_count']} 次") except KeyboardInterrupt: logger.info(f"\n收到中断信号，共完成 {round_count} 轮测试，程序退出。") raise except Exception as e: logger.error(f"测试过程中发生未预期错误: {e}") logger.error(traceback.format_exc()) msg = f"测试过程中发生未预期错误: {e}" if 'executor' in locals(): pass finally: if 'executor' in locals() and executor and hasattr(executor, 'network_watchdog'): logger.info("正在清理网络监控资源...") executor.network_watchdog.shutdown()