# This file contains the main class of droidbot # It can be used after AVD was started, app was installed, and adb had been set up properly # By configuring and creating a droidbot instance, # droidbot will start interacting with Android in AVD like a human import logging import os import sys import pkg_resources import shutil from threading import Timer from .core import PlatformFactory, Platform from .input_manager import InputManager # Import platforms to trigger auto-registration with PlatformFactory # This ensures platforms are registered before create_device is called from . import platforms # noqa: F401 # 从 exceptions 模块导入致命异常（避免循环导入） from .exceptions import FATAL_EXCEPTIONS # noqa: F401 class DroidBot(object): """ The main class of droidbot """ # this is a single instance class instance = None def __init__(self, package_name=None, device_serial=None, is_emulator=False, output_dir=None, policy_name=None, random_input=False, event_count=None, event_interval=None, timeout=None, keep_app=None, keep_env=False, cv_mode=False, debug_mode=False, profiling_method=None, grant_perm=False, enable_accessibility_hard=False, humanoid=None, ignore_ad=False, replay_output=None, enable_guiagent=False, app_name=None, platform="android", pcap_callback=None, enable_app_block=False, **kwargs): """ initiate droidbot with configurations :return: """ # 注意：日志配置现在由统一的 logging_config 模块在入口文件中管理 # 不在这里调用 basicConfig，避免覆盖已有配置 self.logger = logging.getLogger('DroidBot') DroidBot.instance = self self.output_dir = output_dir if output_dir is not None: if not os.path.isdir(output_dir): os.makedirs(output_dir) html_index_path = pkg_resources.resource_filename("DroidBot", "resources/index.html") stylesheets_path = pkg_resources.resource_filename("DroidBot", "resources/stylesheets") target_stylesheets_dir = os.path.join(output_dir, "stylesheets") if os.path.exists(target_stylesheets_dir): shutil.rmtree(target_stylesheets_dir) shutil.copy(html_index_path, output_dir) shutil.copytree(stylesheets_path, target_stylesheets_dir) self.timeout = timeout self.timer = None self.keep_env = keep_env self.keep_app = keep_app self.device = None self.input_manager = None self.enable_accessibility_hard = enable_accessibility_hard self.humanoid = humanoid self.ignore_ad = ignore_ad self.replay_output = replay_output self.enable_guiagent = enable_guiagent self.app_name = app_name self.pcap_callback = pcap_callback self.enable_app_block = enable_app_block self.enabled = True self._timeout_triggered = False self._stopped = False try: # Use PlatformFactory to create device platform_enum = Platform(platform) # Build platform-specific device arguments if platform_enum == Platform.IOS: # iOS device arguments device_kwargs = { 'wda_url': kwargs.get('wda_url', 'http://localhost:8100'), 'bundle_id': package_name, # package_name is bundle_id for iOS 'output_dir': self.output_dir, 'cv_mode': cv_mode, 'udid': device_serial, # device_serial maps to udid for iOS 'debug_mode': debug_mode, } elif platform_enum == Platform.WINDOWS: # Windows device arguments device_kwargs = { 'window_title': self.app_name, # app_name maps to window_title 'exe_path': device_serial, # device_serial maps to exe_path 'steam_game_id': package_name, # package_name maps to steam_game_id 'output_dir': self.output_dir, 'cv_mode': cv_mode, 'debug_mode': debug_mode, } elif platform_enum == Platform.WEB: # Web device arguments device_kwargs = { 'app_path': package_name, # URL作为app_path 'output_dir': self.output_dir, 'browser': kwargs.get('browser', 'chrome'), 'engine': kwargs.get('engine', 'playwright'), 'headless': kwargs.get('headless', False), } else: # Android device arguments device_kwargs = { 'device_serial': device_serial, 'is_emulator': is_emulator, 'output_dir': self.output_dir, 'app_path': package_name, 'cv_mode': cv_mode, 'grant_perm': grant_perm, 'enable_accessibility_hard': self.enable_accessibility_hard, 'humanoid': self.humanoid, 'ignore_ad': ignore_ad, **kwargs, } self.device = PlatformFactory.create_device(platform_enum, **device_kwargs) self.input_manager = InputManager( device=self.device, policy_name=policy_name, random_input=random_input, event_count=event_count, event_interval=event_interval, profiling_method=profiling_method, replay_output=replay_output, enable_guiagent=self.enable_guiagent, app_name=self.app_name, pcap_callback=self.pcap_callback, enable_app_block=self.enable_app_block) except Exception: self.stop() raise def start(self): """ start interacting :return: """ if not self.enabled: return self.logger.info("Starting DroidBot") try: if self.timeout > 0: self.timer = Timer(self.timeout, self._on_timeout) self.timer.start() self.device.set_up() if not self.enabled: return self.device.connect() if not self.enabled: return # self.device.install_app() # app is already installed if not self.enabled: return self.input_manager.start() except Exception: raise finally: self.stop() self.logger.info("DroidBot Stopped") def _on_timeout(self): """Timer 线程只发出停止信号，避免跨线程关闭 Playwright。""" self._timeout_triggered = True self.enabled = False self.logger.warning(f"DroidBot timeout reached ({self.timeout}s), requesting graceful stop") if self.input_manager: self.input_manager.stop() def stop(self): if self._stopped: return self._stopped = True self.enabled = False if self.timer and self.timer.is_alive(): self.timer.cancel() if self.input_manager: self.logger.info("Total steps: %d" % self.input_manager.total_exploring_steps) self.input_manager.stop() if self.device: self.device.disconnect()