""" Web Input Event Implementation Concrete implementation of AbstractInputEvent for Web applications. """ from typing import Optional, Dict, Any, List from selenium.webdriver.common.action_chains import ActionChains from selenium.webdriver.common.keys import Keys from selenium.webdriver.common.by import By from ...core.abstract_input_event import ( AbstractInputEvent, EventType, BaseTouchEvent, BaseLongTouchEvent, BaseSwipeEvent, BaseScrollEvent, BaseSetTextEvent, BaseKeyEvent, BaseKillAppEvent ) class WebTouchEvent(BaseTouchEvent): """ Web 点击事件实现 """ def send(self, device: 'AbstractDevice') -> bool: """ 发送点击事件到 Web 设备 :param device: Web 设备对象 :return: 是否发送成功 """ try: # 如果有视图信息,通过坐标查找元素并点击 if (self.view or (self.x is not None and self.y is not None)) and device.driver: if self.view: bounds = self.view.get('bounds', [[0, 0], [0, 0]]) x = (bounds[0][0] + bounds[1][0]) / 2 y = (bounds[0][1] + bounds[1][1]) / 2 else: # self.x and self.y x = self.x y = self.y if getattr(device, "engine", "selenium") == "playwright": device.driver.mouse.click(x, y) return True # 使用JavaScript执行点击,避免坐标越界问题 js_script = f""" var el = document.elementFromPoint({x}, {y}); if (el) {{ el.click(); return true; }} return false; """ success = device.driver.execute_script(js_script) if not success: device.logger.warning(f"Could not find element at ({x}, {y}) to click.") return success return False except Exception as e: device.logger.error(f"Failed to send Web touch event: {e}") return False class WebLongTouchEvent(BaseLongTouchEvent): """ Web 长按事件实现 """ def send(self, device: 'AbstractDevice') -> bool: """ 发送长按事件到 Web 设备 :param device: Web 设备对象 :return: 是否发送成功 """ try: x, y = None, None if self.view: bounds = self.view.get('bounds', [[0, 0], [0, 0]]) x = (bounds[0][0] + bounds[1][0]) / 2 y = (bounds[0][1] + bounds[1][1]) / 2 elif self.x is not None and self.y is not None: x = self.x y = self.y if x is not None and y is not None and device.driver: if getattr(device, "engine", "selenium") == "playwright": device.driver.mouse.move(x, y) device.driver.mouse.down() import time time.sleep(self.duration / 1000.0) device.driver.mouse.up() return True # 使用JavaScript模拟长按事件 element = device.driver.execute_script(f"return document.elementFromPoint({x}, {y});") if element: js_script = """ var element = arguments[0]; var duration = arguments[1]; var touchStart = new TouchEvent('touchstart', { bubbles: true, cancelable: true, touches: [{ clientX: arguments[2], clientY: arguments[3] }] }); var touchEnd = new TouchEvent('touchend', { bubbles: true, cancelable: true }); element.dispatchEvent(touchStart); setTimeout(function() { element.dispatchEvent(touchEnd); }, duration); """ device.driver.execute_script(js_script, element, self.duration, x, y) return True return False except Exception as e: device.logger.error(f"Failed to send Web long touch event: {e}") return False class WebSwipeEvent(BaseSwipeEvent): """ Web 滑动事件实现 """ def send(self, device: 'AbstractDevice') -> bool: """ 发送滑动事件到 Web 设备 :param device: Web 设备对象 :return: 是否发送成功 """ try: if device.driver: if getattr(device, "engine", "selenium") == "playwright": device.driver.mouse.move(self.start_x, self.start_y) device.driver.mouse.down() import time time.sleep(0.05) # move in steps for swipe gesture device.driver.mouse.move(self.end_x, self.end_y, steps=10) time.sleep(0.1) device.driver.mouse.up() return True # 使用JavaScript模拟滑动事件 js_script = f""" // 创建触摸开始事件 var touchStart = new TouchEvent('touchstart', {{ bubbles: true, cancelable: true, touches: [{{ clientX: {self.start_x}, clientY: {self.start_y} }}] }}); // 创建触摸移动事件 var touchMove = new TouchEvent('touchmove', {{ bubbles: true, cancelable: true, touches: [{{ clientX: {self.end_x}, clientY: {self.end_y} }}] }}); // 创建触摸结束事件 var touchEnd = new TouchEvent('touchend', {{ bubbles: true, cancelable: true }}); // 在body元素上分发事件 document.body.dispatchEvent(touchStart); // 短暂延迟后分发移动事件 setTimeout(function() {{ document.body.dispatchEvent(touchMove); // 再次延迟后分发结束事件 setTimeout(function() {{ document.body.dispatchEvent(touchEnd); }}, 100); }}, 50); """ device.driver.execute_script(js_script) return True return False except Exception as e: device.logger.error(f"Failed to send Web swipe event: {e}") return False class WebScrollEvent(BaseScrollEvent): """ Web 滚动事件实现 """ def send(self, device: 'AbstractDevice') -> bool: """ 发送滚动事件到 Web 设备 :param device: Web 设备对象 :return: 是否发送成功 """ try: if device.driver: if getattr(device, "engine", "selenium") == "playwright": if self.direction == self.DIRECTION_UP: device.driver.keyboard.press("PageUp") elif self.direction == self.DIRECTION_DOWN: device.driver.keyboard.press("PageDown") elif self.direction == self.DIRECTION_LEFT: device.driver.keyboard.press("ArrowLeft") elif self.direction == self.DIRECTION_RIGHT: device.driver.keyboard.press("ArrowRight") return True # 使用 ActionChains 执行滚动操作 actions = ActionChains(device.driver) # 根据方向执行滚动 if self.direction == self.DIRECTION_UP: actions.send_keys(Keys.PAGE_UP) elif self.direction == self.DIRECTION_DOWN: actions.send_keys(Keys.PAGE_DOWN) elif self.direction == self.DIRECTION_LEFT: actions.send_keys(Keys.ARROW_LEFT) elif self.direction == self.DIRECTION_RIGHT: actions.send_keys(Keys.ARROW_RIGHT) actions.perform() return True return False except Exception as e: device.logger.error(f"Failed to send Web scroll event: {e}") return False class WebSetTextEvent(BaseSetTextEvent): """ Web 文本输入事件实现 """ def send(self, device: 'AbstractDevice') -> bool: """ 发送文本输入事件到 Web 设备 :param device: Web 设备对象 :return: 是否发送成功 """ try: if self.view and device.driver: bounds = self.view.get('bounds', [[0, 0], [0, 0]]) x = (bounds[0][0] + bounds[1][0]) / 2 y = (bounds[0][1] + bounds[1][1]) / 2 if getattr(device, "engine", "selenium") == "playwright": escaped_text = self.text.replace('`', '\\`').replace('$', '\\$') js_script = f"""() => {{ var element = document.elementFromPoint({x}, {y}); if (element) {{ element.focus(); element.value = ''; }} }}""" device.driver.evaluate(js_script) device.driver.keyboard.type(self.text) device.driver.evaluate(f"""() => {{ var element = document.elementFromPoint({x}, {y}); if (element) {{ element.dispatchEvent(new Event('input', {{ bubbles: true }})); element.dispatchEvent(new Event('change', {{ bubbles: true }})); }} }}""") return True # 使用JavaScript找到元素并设置文本 element = device.driver.execute_script(f"return document.elementFromPoint({x}, {y});") if element: # 先点击元素以获取焦点 try: device.driver.execute_script("arguments[0].click();", element) except Exception as e: device.logger.warning(f"Failed to focus element before setting text: {e}") # 清空现有文本并输入新文本 device.driver.execute_script(""" var element = arguments[0]; var text = arguments[1]; // 清空现有文本 element.value = ''; // 设置新文本 element.value = text; // 触发input和change事件 var event = new Event('input', { bubbles: true }); element.dispatchEvent(event); var changeEvent = new Event('change', { bubbles: true }); element.dispatchEvent(changeEvent); """, element, self.text) return True return False except Exception as e: device.logger.error(f"Failed to send Web set text event: {e}") return False class WebKeyEvent(BaseKeyEvent): """ Web 按键事件实现 """ def send(self, device: 'AbstractDevice') -> bool: """ 发送按键事件到 Web 设备 :param device: Web 设备对象 :return: 是否发送成功 """ try: if device.driver: if getattr(device, "engine", "selenium") == "playwright": if self.key_name == self.KEY_BACK: device.driver.go_back() return True key_map = { self.KEY_HOME: "Home", self.KEY_ENTER: "Enter", self.KEY_ESCAPE: "Escape" } key = key_map.get(self.key_name, self.key_name) device.driver.keyboard.press(key) return True # 使用 ActionChains 执行按键操作 actions = ActionChains(device.driver) # 映射按键名称到 Keys 常量 key_map = { self.KEY_HOME: Keys.HOME, self.KEY_ENTER: Keys.ENTER, self.KEY_ESCAPE: Keys.ESCAPE } if self.key_name == self.KEY_BACK: # 对于 Web 平台,BACK 通常意味着浏览器后退 device.back() return True key = key_map.get(self.key_name, self.key_name) actions.send_keys(key) actions.perform() return True return False except Exception as e: device.logger.error(f"Failed to send Web key event: {e}") return False class WebKillAppEvent(BaseKillAppEvent): """ Web 应用重置事件 清除浏览器状态(Cookies/Storage)、关闭多余标签页、导航回初始页面。 对应 DroidBot 框架中的 KillAppEvent,在探索循环启动和停滞恢复时调用。 """ def send(self, device: 'AbstractDevice') -> bool: """ 重置 Web 应用状态 :param device: Web 设备对象 :return: 是否发送成功 """ try: if not device.driver: device.logger.warning("WebKillAppEvent: driver not available") return False # 1. 清除浏览器存储状态 try: if getattr(device, "engine", "selenium") == "playwright": device.driver.evaluate("try { localStorage.clear(); } catch(e) {}") device.driver.evaluate("try { sessionStorage.clear(); } catch(e) {}") if hasattr(device, 'playwright_context') and device.playwright_context: device.playwright_context.clear_cookies() else: device.driver.execute_script( "try { localStorage.clear(); } catch(e) {}" "try { sessionStorage.clear(); } catch(e) {}" ) device.driver.delete_all_cookies() except Exception as e: device.logger.warning(f"清除浏览器存储失败(可忽略): {e}") # 2. 关闭所有多余标签页,只保留第一个 try: if getattr(device, "engine", "selenium") == "playwright": if hasattr(device, 'playwright_context') and device.playwright_context: pages = device.playwright_context.pages if len(pages) > 1: for p in pages[1:]: p.close() device.driver = pages[0] device.driver.bring_to_front() device.logger.info(f"关闭了 {len(pages) - 1} 个多余标签页") else: handles = device.driver.window_handles if len(handles) > 1: first_handle = handles[0] for h in handles[1:]: device.driver.switch_to.window(h) device.driver.close() device.driver.switch_to.window(first_handle) device.logger.info(f"关闭了 {len(handles) - 1} 个多余标签页") except Exception as e: device.logger.warning(f"关闭多余标签页失败: {e}") # 3. 导航回初始页面 app_url = device.app_url if app_url: if getattr(device, "engine", "selenium") == "playwright": device.driver.goto(app_url) if hasattr(device, "_wait_for_page_ready"): device._wait_for_page_ready() device._current_url = device.driver.url else: device.driver.get(app_url) if hasattr(device, "_wait_for_page_ready"): device._wait_for_page_ready() device._current_url = device.driver.current_url device.logger.info(f"已导航回初始页面: {app_url}") # 4. 更新窗口句柄缓存 if getattr(device, "engine", "selenium") != "playwright": device._window_handles = set(device.driver.window_handles) return True except Exception as e: device.logger.error(f"WebKillAppEvent 执行失败: {e}") return False