#!/usr/bin/env python3 import os import argparse import sys import time import json import re import logging from pathlib import Path from datetime import datetime from dataclasses import dataclass, asdict # 找到项目根目录 ROOT = Path(__file__).resolve().parent # 确保根目录在Python路径中 if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) # 导入统一的日志配置模块 from logging_config import setup_logging, get_logger from config_loader import load_config as load_runtime_config from utils_android.device_config import ensure_android_wireless_connected, get_android_device_serial # 导入 DroidBot from DroidBot.droidbot import DroidBot from DroidBot.exceptions import ADBException, AppCrashException, ExplorationStuckException, AppNeedUpdateException, AppLaunchErrorException from DroidBot.guiagent_core.decision_maker import GuiAgentDecisionMaker from result_codes import ResultCode, ErrorCategory, ErrorInfo, STUCK_TO_ERROR EXIT_ERROR_GENERAL = ResultCode.ERROR_GENERAL EXIT_SUCCESS = ResultCode.SUCCESS EXIT_ERROR_USER = ResultCode.ERROR_USER EXIT_ADB_ERROR = ResultCode.ADB_ERROR EXIT_NETWORK_ERROR = ResultCode.NETWORK_ERROR EXIT_SIMULATOR_ERROR = ResultCode.SIMULATOR_ERROR EXIT_EXPLORATION_STUCK = ResultCode.EXPLORATION_STUCK EXIT_APP_CRASH_ERROR = ResultCode.APP_CRASH EXIT_APP_NEED_UPDATE = ResultCode.APP_NEED_UPDATE EXIT_APP_LAUNCH_ERROR = ResultCode.APP_LAUNCH_ERROR STUCK_REASON_TO_ERROR_TYPE = {k: v[0].name for k, v in STUCK_TO_ERROR.items()} @dataclass class TaskResult: """单次测试任务的执行结果""" status: str # "SUCCESS" | "FAILED" | "INTERRUPTED" exit_code: int # 退出码 error_reason: str # 错误原因（成功时为空字符串） droidbot_steps: int # DroidBot 执行步数 guiagent_steps: int # GuiAgent 执行步数 total_steps: int # 总步数 duration_seconds: float # 执行耗时（秒） guiagent_message: str = "" # GuiAgent 登录/注册失败消息，用于worker.report汇报 login_count: int = 0 # 进入登录场景的次数 register_count: int = 0 # 进入注册场景的次数 stuck_reason_code: int = 0 # 卡住原因代码 (0=无/其他, 1=登录注册, 2=启动异常, 3=正常, 4-12=其他原因) num_nodes: int = 0 # UTG 节点数 num_reached_activities: int = 0 # UTG 已到达 Activity 数 app_num_total_activities: int = 0 # 应用总 Activity 数 def load_config(json_path: str = None): """从JSON文件加载配置""" try: return load_runtime_config(json_path) except Exception as e: target = json_path or "layered autool config" print(f"[ERROR] 无法从 {target} 加载配置: {e}") raise e class TeeOutput: """将输出同时写入控制台和文件，包括标准输出、标准错误和所有logging输出""" def __init__(self, file_path): self.file_path = file_path self.file = open(file_path, 'w', encoding='utf-8', buffering=1) # 行缓冲 self.stdout = sys.stdout self.stderr = sys.stderr self.log_handler = None def write(self, message): try: self.stdout.write(message) self.file.write(message) except Exception: pass # 忽略写入错误，防止程序崩溃 def flush(self): try: self.stdout.flush() self.file.flush() except Exception: pass def fileno(self): """返回文件描述符，使某些库兼容""" return self.stdout.fileno() def isatty(self): """检查是否为终端""" return self.stdout.isatty() def setup_logging(self): """设置根日志记录器的文件处理器，捕获所有logging模块输出""" import logging self.log_handler = logging.FileHandler(self.file_path, mode='a', encoding='utf-8') self.log_handler.setLevel(logging.DEBUG) # 捕获所有级别 formatter = logging.Formatter('%(asctime)s %(name)-12s %(levelname)-8s %(message)s') self.log_handler.setFormatter(formatter) logging.getLogger().addHandler(self.log_handler) def close(self): """清理资源""" import logging if self.log_handler: logging.getLogger().removeHandler(self.log_handler) self.log_handler.close() self.log_handler = None if self.file and not self.file.closed: self.file.close() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() return False class TaskRunner: """任务执行器：负责构建路径、直接调用 DroidBot 模块并执行""" def __init__(self, config: dict): self.config = config def _build_droidbot_kwargs(self, package_name: str, app_name: str, output_dir: str, **extra_kwargs) -> dict: """构建 DroidBot 初始化所需的参数""" # 统一路径格式为正斜杠 output_dir = output_dir.replace('\\', '/') device_serial = get_android_device_serial(self.config) ensure_android_wireless_connected(self.config, device_serial) mumu_config = self.config.get("mumu", {}) kwargs = { 'package_name': package_name, 'app_name': app_name, 'device_serial': device_serial, 'is_emulator': self.config.get("IS_EMULATOR", False), 'output_dir': output_dir, 'policy_name': self.config.get("POLICY", "memory_guided"), 'random_input': False, 'event_interval': 0, 'timeout': self.config.get("TIMEOUT", 3600), 'event_count': 10000, 'cv_mode': self.config.get("CV_MODE", False), 'debug_mode': False, 'keep_app': self.config.get("KEEP_APP", True), 'keep_env': False, 'profiling_method': None, 'grant_perm': False, 'enable_accessibility_hard': False, 'humanoid': None, 'ignore_ad': False, 'replay_output': None, 'enable_guiagent': self.config.get("ENABLE_GUIAGENT", False), 'platform': "android", 'pcap_callback': extra_kwargs.get('pcap_callback'), 'mumu_manager_path': mumu_config.get("manager_path", ""), 'mumu_vm_index': mumu_config.get("default_vm_index", 2), } return kwargs def run_task(self, package_name: str, app_name: str, output_dir: str, **kwargs) -> TaskResult: """执行单个测试任务，返回 TaskResult""" print(f"\n{'='*50}") print(f"[INFO] 准备测试应用包名: {package_name}") start_time = time.time() ret_code = EXIT_ERROR_GENERAL status = "FAILED" error_reason = "" # 重置全局步数计数器，确保每轮任务独立统计 GuiAgentDecisionMaker.total_steps = 0 # 初始化步数统计，确保 finally 块中能访问到 droidbot_steps = 0 guiagent_steps = 0 total_steps = 0 user_interrupted = False # 追踪用户中断，用于向上传递 os.makedirs(output_dir, exist_ok=True) print(f"[INFO] 输出目录已创建: {output_dir}") # 将 **kwargs (如 pcap_callback) 合并到构建参数中 droidbot_kwargs = self._build_droidbot_kwargs(package_name, app_name, output_dir) enable_app_block = kwargs.pop('enable_app_block', False) droidbot_kwargs.update(kwargs) droidbot_kwargs['enable_app_block'] = enable_app_block if enable_app_block: droidbot_kwargs['timeout'] = self.config.get("BLOCK_TIMEOUT", 1000) print(f"[DEBUG] DroidBot 初始化参数: {droidbot_kwargs}") cmd_log_path = os.path.join(output_dir, "cmd_output.txt") print(f"[INFO] 开始运行DroidBot") print(f"{'='*50}\n") # 设置输出重定向 tee = TeeOutput(cmd_log_path) old_stdout = sys.stdout old_stderr = sys.stderr sys.stdout = tee sys.stderr = tee tee.setup_logging() # 设置日志处理器，捕获所有logging模块输出 try: # 直接创建并运行DroidBot实例 droidbot = DroidBot(**droidbot_kwargs) droidbot.start() ret_code = EXIT_SUCCESS status = "SUCCESS" error_reason = "" # 统计步数 droidbot_steps = droidbot.input_manager.total_exploring_steps guiagent_steps = GuiAgentDecisionMaker.total_steps total_steps = droidbot_steps + guiagent_steps print(f"\n[SUCCESS] 测试任务完成: {app_name}") except KeyboardInterrupt: print(f"\n[WARN] 用户手动中断测试: {package_name}") ret_code = EXIT_ERROR_USER status = "INTERRUPTED" error_reason = "用户手动中断" user_interrupted = True # 标记需要向上传递中断 except ADBException as e: import traceback print(f"\n[ERROR] ADB 断联: {e}") print(traceback.format_exc()) ret_code = EXIT_ADB_ERROR status = "FAILED" error_reason = f"ADB 断联: {e}" except AppCrashException as e: import traceback print(f"\n[ERROR] 应用闪退: {e}") print(traceback.format_exc()) ret_code = EXIT_APP_CRASH_ERROR status = "FAILED" error_reason = f"应用闪退: {e}" except AppNeedUpdateException as e: import traceback print(f"\n[ERROR] 应用需更新: {e}") print(traceback.format_exc()) ret_code = EXIT_APP_NEED_UPDATE status = "FAILED" error_reason = f"应用需更新: {e}" except AppLaunchErrorException as e: import traceback print(f"\n[ERROR] 启动异常: {e}") print(traceback.format_exc()) ret_code = EXIT_APP_LAUNCH_ERROR status = "FAILED" error_reason = f"启动异常: {e}" except ExplorationStuckException as e: import traceback print(f"\n[ERROR] 探索停滞: {e}") print(traceback.format_exc()) ret_code = EXIT_EXPLORATION_STUCK status = "FAILED" error_reason = f"探索停滞: {e}" except Exception as e: import traceback print(f"\n[ERROR] DroidBot 执行异常: {e}") print(traceback.format_exc()) ret_code = EXIT_ERROR_GENERAL status = "FAILED" error_reason = f"执行异常: {e}" finally: # 尝试获取步数（即使异常也尽量获取） try: guiagent_steps = GuiAgentDecisionMaker.total_steps except Exception: pass total_steps = droidbot_steps + guiagent_steps # 读取 GuiAgent 登录/注册失败消息 guiagent_message = "" login_count = 0 register_count = 0 stuck_reason_code = 0 try: if droidbot and droidbot.input_manager and droidbot.input_manager.policy: guiagent_message = getattr(droidbot.input_manager.policy, 'guiagent_message', '') or '' login_count = getattr(droidbot.input_manager.policy, 'login_count', 0) or 0 register_count = getattr(droidbot.input_manager.policy, 'register_count', 0) or 0 stuck_reason_code = getattr(droidbot.input_manager.policy, 'stuck_reason_code', 0) or 0 except Exception: pass # 解析 UTG 文件 num_nodes = 0 num_reached_activities = 0 app_num_total_activities = 0 try: from utils_android.Manager.utg_parser import parse_utg_js utg_path = os.path.join(output_dir, 'utg.js') if os.path.exists(utg_path): utg_result = parse_utg_js(utg_path, package_name) num_nodes = utg_result['num_nodes'] num_reached_activities = utg_result['num_reached_activities'] app_num_total_activities = utg_result['app_num_total_activities'] print(f"[INFO] UTG 节点数: {num_nodes}, 已到达Activity: {num_reached_activities}/{app_num_total_activities}") except Exception as e: print(f"[WARN] 解析 UTG 文件失败: {e}") print(f"[INFO] DroidBot 步数: {droidbot_steps}") print(f"[INFO] GUI Agent 步数: {guiagent_steps}") print(f"[INFO] 总步数: {total_steps}") # 恢复原来的stdout/stderr sys.stdout = old_stdout sys.stderr = old_stderr tee.close() end_time = time.time() duration = end_time - start_time hours, rem = divmod(duration, 3600) minutes, seconds = divmod(rem, 60) time_str = f"{int(hours)}小时 {int(minutes)}分 {int(seconds)}秒" print(f"[INFO] 任务耗时: {time_str}\n") if os.path.exists(output_dir): try: time_file_path = os.path.join(output_dir, "time_cost.txt") with open(time_file_path, "w", encoding="utf-8") as f: f.write(f"Test Duration: {time_str}\n") f.write(f"Start Time: {datetime.fromtimestamp(start_time).strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"End Time: {datetime.fromtimestamp(end_time).strftime('%Y-%m-%d %H:%M:%S')}\n") except Exception as e: print(f"[ERROR] 无法记录耗时文件: {e}") result = TaskResult( status=status, exit_code=ret_code, error_reason=error_reason, droidbot_steps=droidbot_steps, guiagent_steps=guiagent_steps, total_steps=total_steps, duration_seconds=duration, guiagent_message=guiagent_message, login_count=login_count, register_count=register_count, stuck_reason_code=stuck_reason_code, num_nodes=num_nodes, num_reached_activities=num_reached_activities, app_num_total_activities=app_num_total_activities ) # 如果是用户中断，返回结果后重新抛出异常，让上层 batch_run 能够捕获并退出 if user_interrupted: raise KeyboardInterrupt(result) return result def main(): logger = get_logger(__name__) logger.info("开始测试任务") package = "com.google.android.youtube" app_name = "YouTube" config = load_config() task_runner = TaskRunner(config) output_dir = "test_output" result = task_runner.run_task(package, app_name, output_dir) logger.info(f"测试完成: status={result.status}, total_steps={result.total_steps}") return result if __name__ == "__main__": setup_logging(level=logging.INFO, enable_file_handler=False) # from pyinstrument import Profiler # profiler = Profiler() # profiler.start() main() # profiler.stop() # profiler.write_html("report.html")