#!/usr/bin/env python3 """ Web 端 DroidBot 自动化批量测试脚本 从CSV文件加载URL列表，逐个执行DroidBot探索测试。 使用示例: # 使用全部默认参数运行 python web_test.py # 指定URL列表 python web_test.py -url_list doc/web_urls.csv # 覆盖部分默认值 python web_test.py -url_list urls.csv -duration 300 -headless """ import os import sys import csv import time import json import logging import argparse from pathlib import Path from datetime import datetime from dataclasses import dataclass, asdict # 找到项目根目录 ROOT = Path(__file__).resolve().parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from logging_config import setup_logging, get_logger from DroidBot.droidbot import DroidBot from DroidBot.exceptions import ( AppCrashException, ExplorationStuckException, AppNeedUpdateException, AppLaunchErrorException ) from DroidBot.guiagent_core.decision_maker import GuiAgentDecisionMaker # ==================== 结果码 ==================== EXIT_SUCCESS = 0 EXIT_ERROR_GENERAL = 1 EXIT_ERROR_USER = 2 EXIT_EXPLORATION_STUCK = 10 # ==================== 数据类 ==================== @dataclass class WebTaskResult: """单次Web测试任务的执行结果""" url: str app_name: str status: str # "SUCCESS" | "FAILED" | "INTERRUPTED" exit_code: int error_reason: str droidbot_steps: int total_steps: int duration_seconds: float num_nodes: int = 0 num_reached_pages: int = 0 # ==================== 统计类 ==================== class WebBatchStatistics: """Web批量测试统计""" def __init__(self, output_dir: str): self.output_dir = output_dir self.results: list = [] self.start_time = time.time() self._summary_path = os.path.join(output_dir, "batch_summary.csv") self._history_path = os.path.join(output_dir, "test_history.json") def add_result(self, result: WebTaskResult): self.results.append(result) self._save_summary() def _save_summary(self): """保存CSV汇总""" try: with open(self._summary_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=[ 'url', 'app_name', 'status', 'error_reason', 'droidbot_steps', 'total_steps', 'duration_seconds', 'num_nodes', 'num_reached_pages' ]) writer.writeheader() for r in self.results: writer.writerow({ 'url': r.url, 'app_name': r.app_name, 'status': r.status, 'error_reason': r.error_reason, 'droidbot_steps': r.droidbot_steps, 'total_steps': r.total_steps, 'duration_seconds': round(r.duration_seconds, 1), 'num_nodes': r.num_nodes, 'num_reached_pages': r.num_reached_pages, }) except Exception as e: print(f"[WARN] 保存汇总CSV失败: {e}") def load_history(self) -> set: """加载已成功测试的URL集合（智能调度用）""" succeeded = set() if os.path.exists(self._history_path): try: with open(self._history_path, 'r') as f: history = json.load(f) for entry in history: if entry.get('status') == 'SUCCESS': succeeded.add(entry.get('url', '')) except Exception: pass return succeeded def save_history(self): """保存测试历史""" try: with open(self._history_path, 'w') as f: json.dump([asdict(r) for r in self.results], f, indent=2, ensure_ascii=False) except Exception as e: print(f"[WARN] 保存测试历史失败: {e}") def print_summary(self): """打印最终统计""" total = len(self.results) success = sum(1 for r in self.results if r.status == 'SUCCESS') failed = sum(1 for r in self.results if r.status == 'FAILED') elapsed = time.time() - self.start_time hours, rem = divmod(elapsed, 3600) minutes, seconds = divmod(rem, 60) print(f"\n{'='*60}") print(f" Web 批量测试完成") print(f"{'='*60}") print(f" 总计: {total} 成功: {success} 失败: {failed}") print(f" 总耗时: {int(hours)}h {int(minutes)}m {int(seconds)}s") print(f" 汇总文件: {self._summary_path}") print(f"{'='*60}\n") # ==================== 测试执行器 ==================== class WebTestRunner: """Web测试执行器：直接调用DroidBot模块""" def __init__(self, duration: int, headless: bool, output_base: str, engine: str = 'playwright', browser: str = 'chrome'): self.duration = duration self.headless = headless self.output_base = output_base self.engine = engine self.browser = browser self.logger = get_logger('WebTestRunner') def run_test(self, url: str, app_name: str) -> WebTaskResult: """执行单个URL的DroidBot测试""" # 构建输出目录（基于域名 + 时间戳） from urllib.parse import urlparse domain = urlparse(url).netloc or "unknown" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_dir = os.path.join(self.output_base, f"{domain}_{timestamp}") os.makedirs(output_dir, exist_ok=True) self.logger.info(f"\n{'='*50}") self.logger.info(f"开始测试: {app_name} ({url})") self.logger.info(f"输出目录: {output_dir}") self.logger.info(f"{'='*50}") start_time = time.time() status = "FAILED" exit_code = EXIT_ERROR_GENERAL error_reason = "" droidbot_steps = 0 total_steps = 0 num_nodes = 0 num_reached_pages = 0 # 重置全局步数计数器 GuiAgentDecisionMaker.total_steps = 0 droidbot = None try: droidbot = DroidBot( package_name=url, app_name=app_name, device_serial=None, is_emulator=False, output_dir=output_dir, policy_name="memory_guided", random_input=False, event_interval=0.3, timeout=self.duration, event_count=10000, cv_mode=False, debug_mode=False, 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=False, platform="web", browser=self.browser, engine=self.engine, headless=self.headless, ) droidbot.start() exit_code = EXIT_SUCCESS status = "SUCCESS" except KeyboardInterrupt: status = "INTERRUPTED" exit_code = EXIT_ERROR_USER error_reason = "用户手动中断" raise # 向上传播 except ExplorationStuckException as e: status = "FAILED" exit_code = EXIT_EXPLORATION_STUCK error_reason = f"探索停滞: {e}" except Exception as e: import traceback self.logger.error(f"DroidBot执行异常: {e}") self.logger.error(traceback.format_exc()) status = "FAILED" error_reason = f"执行异常: {e}" finally: # 尝试获取步数 try: if droidbot and droidbot.input_manager: droidbot_steps = droidbot.input_manager.total_exploring_steps except Exception: pass total_steps = droidbot_steps + GuiAgentDecisionMaker.total_steps # 尝试获取UTG节点数 try: if droidbot and droidbot.input_manager and droidbot.input_manager.policy: utg = getattr(droidbot.input_manager.policy, 'utg', None) if utg: num_nodes = len(utg.G.nodes()) num_reached_pages = len(utg.reached_activities) except Exception: pass duration = time.time() - start_time self.logger.info(f"测试完成: {app_name} | 状态={status} | 步数={droidbot_steps} | 节点={num_nodes} | 耗时={duration:.0f}s") return WebTaskResult( url=url, app_name=app_name, status=status, exit_code=exit_code, error_reason=error_reason, droidbot_steps=droidbot_steps, total_steps=total_steps, duration_seconds=duration, num_nodes=num_nodes, num_reached_pages=num_reached_pages, ) # ==================== URL加载 ==================== def load_urls(csv_path: str) -> list: """ 从CSV加载URL列表 支持格式: - AppName,URL（标准格式） - 每行一个URL（无表头） """ urls = [] try: with open(csv_path, 'r', encoding='utf-8') as f: reader = csv.reader(f) header = next(reader, None) # 判断是否有表头 if header and any(h.lower().startswith('http') for h in header): # 无表头，第一行就是数据 if len(header) >= 2: urls.append((header[0].strip(), header[1].strip())) else: urls.append(("Unknown", header[0].strip())) for row in reader: if not row or not row[0].strip(): continue if len(row) >= 2: app_name = row[0].strip() url = row[1].strip() else: app_name = row[0].strip() url = row[0].strip() # 确保是有效URL if url.startswith('http'): urls.append((app_name, url)) except FileNotFoundError: print(f"[ERROR] URL列表文件不存在: {csv_path}") sys.exit(1) except Exception as e: print(f"[ERROR] 读取URL列表失败: {e}") sys.exit(1) return urls # ==================== 主入口 ==================== def parse_args(): parser = argparse.ArgumentParser( description="Web端DroidBot自动化批量测试", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例: python web_test.py # 全部默认参数 python web_test.py -url_list urls.csv # 指定URL列表 python web_test.py -duration 300 -headless # 无头模式，每URL 5分钟 """ ) parser.add_argument('-url_list', type=str, default='output/web_urls.csv', help='URL列表CSV文件 (默认: doc/web_urls.csv)') parser.add_argument('-duration', type=int, default=3600, help='每个URL测试时长，秒 (默认: 3600)') parser.add_argument('-headless', action='store_true', default=False, help='使用无头模式 (默认: False)') parser.add_argument('-output_dir', type=str, default='output/web_test', help='输出目录 (默认: output/web_test)') parser.add_argument('-engine', type=str, choices=['playwright', 'selenium'], default='playwright', help='Web自动化引擎 (默认: playwright)') parser.add_argument('-browser', type=str, choices=['chrome', 'lightpanda'], default='chrome', help='使用的浏览器 (默认: chrome)') parser.add_argument('-batch_count', type=int, default=0, help='测试URL数量上限，0=全部 (默认: 0)') return parser.parse_args() def main(): args = parse_args() setup_logging(level=logging.INFO, enable_file_handler=False) logger = get_logger(__name__) # 加载URL列表 csv_path = os.path.join(ROOT, args.url_list) if not os.path.isabs(args.url_list) else args.url_list all_urls = load_urls(csv_path) logger.info(f"从 {csv_path} 加载了 {len(all_urls)} 个URL") if not all_urls: logger.error("URL列表为空，退出") sys.exit(1) # 初始化输出目录和统计 output_dir = os.path.join(ROOT, args.output_dir) if not os.path.isabs(args.output_dir) else args.output_dir os.makedirs(output_dir, exist_ok=True) stats = WebBatchStatistics(output_dir) # 智能调度：跳过已成功的URL succeeded_urls = stats.load_history() pending_urls = [(name, url) for name, url in all_urls if url not in succeeded_urls] if len(pending_urls) < len(all_urls): logger.info(f"智能调度: 跳过 {len(all_urls) - len(pending_urls)} 个已成功的URL") # 限制数量 if args.batch_count > 0: pending_urls = pending_urls[:args.batch_count] logger.info(f"待测试URL: {len(pending_urls)} 个") # 执行测试 runner = WebTestRunner( duration=args.duration, headless=args.headless, output_base=output_dir, engine=args.engine, browser=args.browser, ) try: for idx, (app_name, url) in enumerate(pending_urls, 1): logger.info(f"\n[{idx}/{len(pending_urls)}] 开始测试: {app_name}") try: result = runner.run_test(url, app_name) stats.add_result(result) except KeyboardInterrupt: logger.warning("用户中断，保存当前进度...") break except Exception as e: logger.error(f"测试 {app_name} 发生未捕获异常: {e}") stats.add_result(WebTaskResult( url=url, app_name=app_name, status="FAILED", exit_code=EXIT_ERROR_GENERAL, error_reason=str(e), droidbot_steps=0, total_steps=0, duration_seconds=0, )) finally: stats.save_history() stats.print_summary() if __name__ == "__main__": main()