815 lines
33 KiB
Python
815 lines
33 KiB
Python
"""
|
||
Web Device Implementation
|
||
Concrete implementation of AbstractDevice for Web applications.
|
||
"""
|
||
import logging
|
||
import os
|
||
import time
|
||
from typing import Optional, Dict, Any, List
|
||
|
||
from selenium import webdriver
|
||
from selenium.webdriver.chrome.options import Options
|
||
from webdriver_manager.chrome import ChromeDriverManager
|
||
from selenium.webdriver.chrome.service import Service as ChromeService
|
||
|
||
from ...core.abstract_device import AbstractDevice
|
||
|
||
|
||
class WebDevice(AbstractDevice):
|
||
"""
|
||
Web 设备的具体实现
|
||
|
||
继承自 AbstractDevice，使用 Selenium WebDriver 实现所有 Web 特定的设备操作。
|
||
"""
|
||
|
||
def __init__(self, app_path: Optional[str] = None, output_dir: Optional[str] = None,
|
||
browser: str = "chrome", headless: bool = False, **kwargs):
|
||
"""
|
||
初始化 Web 设备连接
|
||
|
||
:param app_path: Web 应用的 URL 或本地 HTML 文件路径
|
||
:param output_dir: 输出目录
|
||
:param browser: 浏览器类型，支持 "chrome" 等
|
||
:param headless: 是否使用无头模式
|
||
:param kwargs: 其他参数
|
||
"""
|
||
super().__init__(output_dir=output_dir)
|
||
|
||
self.browser = browser.lower()
|
||
self.headless = headless
|
||
self.engine = kwargs.get('engine', 'playwright')
|
||
self.driver = None # Selenium WebDriver实例 或 Playwright Page实例
|
||
self.playwright = None
|
||
self.playwright_browser = None
|
||
self.playwright_context = None
|
||
self._current_url = None # 当前页面URL缓存
|
||
self._window_handles = set() # 记录已知的窗口句柄
|
||
self._last_sync_time = 0 # P2: 窗口句柄同步时间戳缓存
|
||
self._last_state = None
|
||
self._view_limit = int(kwargs.get('view_limit', 250))
|
||
|
||
# 设备信息
|
||
self.display_info = None
|
||
|
||
# 初始化WebApp实例（与AndroidDevice保持一致的命名方式）
|
||
self.app = None
|
||
self._app = None
|
||
if app_path:
|
||
from .web_app import WebApp
|
||
self.app = WebApp(app_path, output_dir=output_dir)
|
||
self._app = self.app # 与AndroidDevice保持一致，使用_app属性
|
||
self.logger.info(f"初始化WebApp: {app_path}")
|
||
|
||
@property
|
||
def app_url(self) -> Optional[str]:
|
||
"""获取 Web 应用的 URL"""
|
||
if self._app:
|
||
return self._app.app_url
|
||
return None
|
||
|
||
|
||
# ==================== 平台信息 ====================
|
||
|
||
def get_platform_name(self) -> str:
|
||
return "web"
|
||
|
||
|
||
# ==================== 连接管理 ====================
|
||
|
||
def set_up(self) -> None:
|
||
"""设置WebDriver或Playwright"""
|
||
self.logger.info(f"Setting up Web engine ({self.engine})...")
|
||
|
||
if self.engine == "playwright":
|
||
from playwright.sync_api import sync_playwright
|
||
self.playwright = sync_playwright().start()
|
||
|
||
if self.browser == "lightpanda":
|
||
self.logger.info("Connecting to local Lightpanda via CDP (ws://127.0.0.1:9222)...")
|
||
try:
|
||
self.playwright_browser = self.playwright.chromium.connect_over_cdp("ws://127.0.0.1:9222")
|
||
self.playwright_context = self.playwright_browser.contexts[0] if self.playwright_browser.contexts else self.playwright_browser.new_context()
|
||
self.driver = self.playwright_context.pages[0] if self.playwright_context.pages else self.playwright_context.new_page()
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to connect to Lightpanda: {e}")
|
||
raise
|
||
else:
|
||
self.logger.info("Launching Playwright Chrome...")
|
||
self.playwright_browser = self.playwright.chromium.launch(headless=self.headless)
|
||
self.playwright_context = self.playwright_browser.new_context(
|
||
viewport={'width': 1920, 'height': 1080},
|
||
ignore_https_errors=True
|
||
)
|
||
self.driver = self.playwright_context.new_page()
|
||
else:
|
||
# 配置浏览器选项 (Selenium logic)
|
||
if self.browser == "chrome":
|
||
options = Options()
|
||
if self.headless:
|
||
options.add_argument("--headless")
|
||
# 窗口配置
|
||
options.add_argument("--window-size=1920,1080")
|
||
# 性能和兼容性配置
|
||
options.add_argument("--no-sandbox")
|
||
options.add_argument("--disable-dev-shm-usage")
|
||
options.add_argument("--disable-gpu") # 禁用GPU加速
|
||
# 用户体验配置
|
||
options.add_argument("--disable-infobars") # 禁用信息栏
|
||
options.add_argument("--disable-extensions") # 禁用扩展
|
||
# SSL/证书配置
|
||
options.add_argument("--ignore-certificate-errors") # 忽略证书错误
|
||
options.add_argument("--ignore-ssl-errors") # 忽略SSL错误
|
||
# 自动化检测规避
|
||
options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
||
options.add_experimental_option("useAutomationExtension", False)
|
||
|
||
# 初始化WebDriver - 优先使用本地chromedriver，避免每次检查更新
|
||
try:
|
||
# 尝试直接使用本地chromedriver（系统PATH中或默认位置）
|
||
self.driver = webdriver.Chrome(options=options)
|
||
self.logger.info("使用本地chromedriver")
|
||
except Exception as e:
|
||
# 如果本地没有找到，再使用webdriver_manager下载
|
||
self.logger.warning(f"本地chromedriver未找到，尝试下载: {e}")
|
||
self.driver = webdriver.Chrome(
|
||
service=ChromeService(ChromeDriverManager(cache_valid_range=365).install()),
|
||
options=options
|
||
)
|
||
else:
|
||
raise NotImplementedError(f"Browser {self.browser} is not supported yet")
|
||
|
||
def _make_state_tag(self) -> str:
|
||
"""生成高精度状态标签，避免同秒内截图文件名冲突。"""
|
||
from datetime import datetime
|
||
return datetime.now().strftime("%Y-%m-%d_%H%M%S_%f")
|
||
|
||
def _wait_for_page_ready(self, timeout_ms: int = 5000) -> None:
|
||
"""在截图和控件提取前等待页面达到可操作状态。"""
|
||
if not self.driver:
|
||
return
|
||
|
||
try:
|
||
if self.engine == "playwright":
|
||
self.driver.wait_for_load_state("domcontentloaded", timeout=timeout_ms)
|
||
try:
|
||
self.driver.wait_for_load_state("load", timeout=timeout_ms)
|
||
except Exception:
|
||
# 某些站点会持续请求资源，load 超时不应阻塞探索。
|
||
pass
|
||
self.driver.wait_for_timeout(200)
|
||
else:
|
||
deadline = time.time() + timeout_ms / 1000.0
|
||
while time.time() < deadline:
|
||
try:
|
||
ready_state = self.driver.execute_script("return document.readyState")
|
||
if ready_state in ("interactive", "complete"):
|
||
break
|
||
except Exception:
|
||
break
|
||
time.sleep(0.1)
|
||
except Exception as e:
|
||
self.logger.debug(f"wait_for_page_ready skipped: {e}")
|
||
|
||
def get_views(self) -> List[Dict[str, Any]]:
|
||
"""从当前 DOM 提取可交互控件，统一转换为 ViewDict。"""
|
||
if not self.driver:
|
||
return []
|
||
|
||
self._wait_for_page_ready()
|
||
|
||
script = f"""
|
||
() => {{
|
||
const MAX_VIEWS = {self._view_limit};
|
||
const viewportW = window.innerWidth || document.documentElement.clientWidth || 1920;
|
||
const viewportH = window.innerHeight || document.documentElement.clientHeight || 1080;
|
||
const allElements = Array.from(document.querySelectorAll('body *'));
|
||
|
||
const normalizeText = (value) => (value || '')
|
||
.replace(/\\s+/g, ' ')
|
||
.trim()
|
||
.slice(0, 200);
|
||
|
||
const isVisible = (el) => {{
|
||
if (!el || !el.isConnected) return false;
|
||
const style = window.getComputedStyle(el);
|
||
if (!style) return false;
|
||
if (style.display === 'none' || style.visibility === 'hidden' || style.pointerEvents === 'none') return false;
|
||
if (Number(style.opacity || '1') === 0) return false;
|
||
const rect = el.getBoundingClientRect();
|
||
if (!rect || rect.width < 4 || rect.height < 4) return false;
|
||
if (rect.bottom <= 0 || rect.right <= 0 || rect.top >= viewportH || rect.left >= viewportW) return false;
|
||
return true;
|
||
}};
|
||
|
||
const isEnabled = (el) => !el.disabled && el.getAttribute('aria-disabled') !== 'true';
|
||
|
||
const isEditable = (el) => {{
|
||
if (!el) return false;
|
||
if (el.isContentEditable) return true;
|
||
const tag = el.tagName;
|
||
if (tag === 'TEXTAREA' || tag === 'SELECT') return true;
|
||
if (tag === 'INPUT') {{
|
||
const type = (el.getAttribute('type') || 'text').toLowerCase();
|
||
return !['button', 'checkbox', 'color', 'file', 'hidden', 'image', 'radio', 'range', 'reset', 'submit'].includes(type);
|
||
}}
|
||
return false;
|
||
}};
|
||
|
||
const isScrollable = (el) => {{
|
||
if (!el) return false;
|
||
const style = window.getComputedStyle(el);
|
||
if (!style) return false;
|
||
const overflowY = style.overflowY || '';
|
||
const overflowX = style.overflowX || '';
|
||
const scrollY = el.scrollHeight - el.clientHeight > 24 && ['auto', 'scroll', 'overlay'].includes(overflowY);
|
||
const scrollX = el.scrollWidth - el.clientWidth > 24 && ['auto', 'scroll', 'overlay'].includes(overflowX);
|
||
return scrollY || scrollX;
|
||
}};
|
||
|
||
const isClickable = (el, editable) => {{
|
||
if (!el) return false;
|
||
if (editable) return true;
|
||
const tag = el.tagName;
|
||
const role = (el.getAttribute('role') || '').toLowerCase();
|
||
const type = (el.getAttribute('type') || '').toLowerCase();
|
||
const style = window.getComputedStyle(el);
|
||
return Boolean(
|
||
typeof el.onclick === 'function' ||
|
||
el.hasAttribute('onclick') ||
|
||
el.hasAttribute('ng-click') ||
|
||
el.hasAttribute('v-on:click') ||
|
||
el.hasAttribute('@click') ||
|
||
(tag === 'A' && el.getAttribute('href')) ||
|
||
['BUTTON', 'SUMMARY', 'LABEL'].includes(tag) ||
|
||
['button', 'link', 'tab', 'menuitem', 'checkbox', 'radio', 'switch', 'option'].includes(role) ||
|
||
(tag === 'INPUT' && ['button', 'checkbox', 'radio', 'submit'].includes(type)) ||
|
||
(style && style.cursor === 'pointer') ||
|
||
el.tabIndex >= 0
|
||
);
|
||
}};
|
||
|
||
const getText = (el) => {{
|
||
if (!el) return '';
|
||
const tag = el.tagName;
|
||
if (tag === 'INPUT' || tag === 'TEXTAREA' || tag === 'SELECT') {{
|
||
return normalizeText(el.value || el.getAttribute('placeholder') || el.getAttribute('aria-label'));
|
||
}}
|
||
return normalizeText(el.innerText || el.textContent || el.getAttribute('aria-label') || el.getAttribute('title'));
|
||
}};
|
||
|
||
const getDescription = (el) => normalizeText(
|
||
el.getAttribute('aria-label') ||
|
||
el.getAttribute('title') ||
|
||
el.getAttribute('placeholder') ||
|
||
''
|
||
);
|
||
|
||
const decorativeTags = new Set(['path', 'svg', 'g', 'use', 'circle', 'rect', 'polygon', 'line', 'polyline', 'ellipse']);
|
||
const semanticTags = new Set(['a', 'button', 'input', 'textarea', 'select', 'summary', 'label', 'option']);
|
||
|
||
const toBounds = (rect) => {{
|
||
const left = Math.max(0, Math.floor(rect.left));
|
||
const top = Math.max(0, Math.floor(rect.top));
|
||
const right = Math.min(viewportW, Math.ceil(rect.right));
|
||
const bottom = Math.min(viewportH, Math.ceil(rect.bottom));
|
||
return [[left, top], [right, bottom]];
|
||
}};
|
||
|
||
const candidates = [];
|
||
for (const el of allElements) {{
|
||
if (!isVisible(el)) continue;
|
||
|
||
const editable = isEditable(el);
|
||
const scrollable = isScrollable(el);
|
||
const clickable = isClickable(el, editable);
|
||
if (!clickable && !editable && !scrollable) continue;
|
||
|
||
const rect = el.getBoundingClientRect();
|
||
const area = Math.max(1, rect.width * rect.height);
|
||
if (area >= viewportW * viewportH * 0.98 && !editable && !scrollable) continue;
|
||
const className = (el.tagName || 'div').toLowerCase();
|
||
const text = getText(el);
|
||
const contentDescription = getDescription(el);
|
||
if (decorativeTags.has(className) && !text && !contentDescription && area < 2000) continue;
|
||
const semantic = semanticTags.has(className) ? 1 : 0;
|
||
const hasText = text || contentDescription ? 1 : 0;
|
||
|
||
candidates.push({{
|
||
element: el,
|
||
bounds: toBounds(rect),
|
||
text,
|
||
content_description: contentDescription,
|
||
visible: true,
|
||
enabled: isEnabled(el),
|
||
clickable,
|
||
editable,
|
||
scrollable,
|
||
checkable: ['checkbox', 'radio', 'switch'].includes((el.getAttribute('role') || '').toLowerCase()) ||
|
||
(el.tagName === 'INPUT' && ['checkbox', 'radio'].includes((el.getAttribute('type') || '').toLowerCase())),
|
||
checked: Boolean(el.checked),
|
||
selected: Boolean(el.selected || el.getAttribute('aria-selected') === 'true'),
|
||
long_clickable: false,
|
||
children: [],
|
||
parent: -1,
|
||
resource_id: el.id || '',
|
||
class_name: className,
|
||
source: 'dom',
|
||
area,
|
||
priority: editable ? 5 : (semantic ? 4 : (hasText ? 3 : (clickable ? 2 : 1)))
|
||
}});
|
||
}}
|
||
|
||
candidates.sort((a, b) => {{
|
||
if (b.priority !== a.priority) return b.priority - a.priority;
|
||
return a.area - b.area;
|
||
}});
|
||
|
||
const selected = [];
|
||
const seen = new Set();
|
||
for (const candidate of candidates) {{
|
||
const key = `${{candidate.class_name}}|${{candidate.bounds[0][0]}},${{candidate.bounds[0][1]}},${{candidate.bounds[1][0]}},${{candidate.bounds[1][1]}}|${{candidate.text}}`;
|
||
if (seen.has(key)) continue;
|
||
seen.add(key);
|
||
selected.push(candidate);
|
||
if (selected.length >= MAX_VIEWS) break;
|
||
}}
|
||
|
||
const indexByElement = new Map();
|
||
selected.forEach((candidate, index) => {{
|
||
indexByElement.set(candidate.element, index);
|
||
}});
|
||
|
||
selected.forEach((candidate, index) => {{
|
||
let parent = candidate.element.parentElement;
|
||
while (parent) {{
|
||
if (indexByElement.has(parent)) {{
|
||
const parentIndex = indexByElement.get(parent);
|
||
candidate.parent = parentIndex;
|
||
selected[parentIndex].children.push(index);
|
||
break;
|
||
}}
|
||
parent = parent.parentElement;
|
||
}}
|
||
}});
|
||
|
||
return selected.map((candidate, index) => {{
|
||
delete candidate.element;
|
||
delete candidate.area;
|
||
delete candidate.priority;
|
||
candidate.temp_id = index;
|
||
return candidate;
|
||
}});
|
||
}}
|
||
"""
|
||
|
||
try:
|
||
if self.engine == "playwright":
|
||
views = self.driver.evaluate(script)
|
||
else:
|
||
views = self.driver.execute_script(f"return ({script})();")
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to extract Web views: {e}")
|
||
return []
|
||
|
||
if not isinstance(views, list):
|
||
return []
|
||
self.logger.info(f"Extracted {len(views)} Web views")
|
||
return views
|
||
|
||
def connect(self) -> bool:
|
||
"""连接到Web应用"""
|
||
try:
|
||
if self._app:
|
||
self.logger.info(f"Connecting to Web application: {self._app.app_url}")
|
||
if self.driver is None:
|
||
self.set_up()
|
||
|
||
if self.engine == "playwright":
|
||
self.driver.goto(self._app.app_url)
|
||
self._wait_for_page_ready()
|
||
self._current_url = self.driver.url
|
||
else:
|
||
self.driver.get(self._app.app_url)
|
||
self._wait_for_page_ready()
|
||
self._current_url = self.driver.current_url
|
||
self._window_handles = set(self.driver.window_handles)
|
||
self.connected = True
|
||
self.logger.info(f"Connected to Web application: {self._current_url}")
|
||
return True
|
||
else:
|
||
self.logger.error("No Web application URL specified")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to connect to Web application: {e}")
|
||
return False
|
||
|
||
def disconnect(self) -> None:
|
||
"""断开连接"""
|
||
self.logger.info("Disconnecting Web engine...")
|
||
self.connected = False
|
||
if self.engine == "playwright":
|
||
if self.driver:
|
||
try:
|
||
if not self.driver.is_closed():
|
||
self.driver.close()
|
||
except Exception as e:
|
||
self.logger.debug(f"Close Playwright page skipped: {e}")
|
||
if self.playwright_context:
|
||
try:
|
||
self.playwright_context.close()
|
||
except Exception as e:
|
||
self.logger.debug(f"Close Playwright context skipped: {e}")
|
||
if self.playwright_browser:
|
||
try:
|
||
self.playwright_browser.close()
|
||
except Exception as e:
|
||
self.logger.debug(f"Close Playwright browser skipped: {e}")
|
||
if self.playwright:
|
||
try:
|
||
self.playwright.stop()
|
||
except Exception as e:
|
||
self.logger.debug(f"Stop Playwright skipped: {e}")
|
||
self.driver = None
|
||
self.playwright_context = None
|
||
self.playwright_browser = None
|
||
self.playwright = None
|
||
else:
|
||
if self.driver:
|
||
try:
|
||
self.driver.quit()
|
||
except Exception as e:
|
||
self.logger.debug(f"Quit Selenium driver skipped: {e}")
|
||
self.driver = None
|
||
|
||
def tear_down(self) -> None:
|
||
"""清理WebDriver资源"""
|
||
self.disconnect()
|
||
|
||
def check_connectivity(self) -> bool:
|
||
"""检查WebDriver连接状态"""
|
||
try:
|
||
if self.driver:
|
||
if self.engine == "playwright":
|
||
if self.driver.is_closed():
|
||
raise Exception("Playwright page is closed")
|
||
self.driver.title()
|
||
else:
|
||
self.driver.title # 尝试获取页面标题，如果失败则表示连接已断开
|
||
self.connected = True
|
||
return True
|
||
else:
|
||
self.connected = False
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"WebDriver connectivity check failed: {e}")
|
||
self.connected = False
|
||
return False
|
||
|
||
# ==================== 状态获取 ====================
|
||
|
||
def get_current_state(self) -> 'WebDeviceState':
|
||
"""获取当前Web页面状态"""
|
||
from .web_device_state import WebDeviceState
|
||
|
||
self.logger.debug("Getting current Web page state...")
|
||
try:
|
||
tag = self._make_state_tag()
|
||
views = self.get_views()
|
||
screenshot_path = self.take_screenshot(tag=tag)
|
||
current_state = WebDeviceState(
|
||
self,
|
||
views=views,
|
||
tag=tag,
|
||
screenshot_path=screenshot_path
|
||
)
|
||
return current_state
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to get current Web page state: {e}")
|
||
return None
|
||
|
||
def get_display_info(self, refresh: bool = False) -> Dict[str, Any]:
|
||
"""获取显示信息"""
|
||
if self.display_info is None or refresh:
|
||
if self.driver:
|
||
if self.engine == "playwright":
|
||
viewport = self.driver.viewport_size
|
||
if viewport:
|
||
width = viewport['width']
|
||
height = viewport['height']
|
||
else:
|
||
width, height = 1920, 1080
|
||
else:
|
||
window_size = self.driver.get_window_size()
|
||
width = window_size["width"]
|
||
height = window_size["height"]
|
||
self.display_info = {
|
||
"width": width,
|
||
"height": height,
|
||
"density": 1.0 # Web没有物理密度概念，使用1.0
|
||
}
|
||
else:
|
||
self.display_info = {
|
||
"width": 1920,
|
||
"height": 1080,
|
||
"density": 1.0
|
||
}
|
||
return self.display_info
|
||
|
||
# ==================== 屏幕操作 ====================
|
||
|
||
def take_screenshot(self, path: str = None, tag: str = None) -> str:
|
||
"""截取Web页面截图
|
||
|
||
P4: 支持 tag 参数，与state关联便于去重
|
||
"""
|
||
if not self.driver:
|
||
self.logger.error("WebDriver not initialized, cannot take screenshot")
|
||
return None
|
||
|
||
if self.output_dir is None and path is None:
|
||
return None
|
||
|
||
if tag is None:
|
||
from datetime import datetime
|
||
tag = datetime.now().strftime("%Y-%m-%d_%H%M%S")
|
||
local_image_dir = os.path.join(self.output_dir, "temp") if self.output_dir else "/tmp"
|
||
if not os.path.exists(local_image_dir):
|
||
os.makedirs(local_image_dir)
|
||
|
||
if path is None:
|
||
local_image_path = os.path.join(local_image_dir, f"web_screen_{tag}.png")
|
||
else:
|
||
local_image_path = path
|
||
|
||
parent_dir = os.path.dirname(local_image_path)
|
||
if parent_dir and not os.path.exists(parent_dir):
|
||
os.makedirs(parent_dir)
|
||
|
||
last_error = None
|
||
for attempt in range(2):
|
||
try:
|
||
self._wait_for_page_ready(timeout_ms=2000)
|
||
if self.engine == "playwright":
|
||
self.driver.screenshot(path=local_image_path)
|
||
else:
|
||
success = self.driver.save_screenshot(local_image_path)
|
||
if not success:
|
||
raise RuntimeError("save_screenshot returned False")
|
||
if os.path.exists(local_image_path) and os.path.getsize(local_image_path) > 0:
|
||
self.logger.info(f"Screenshot saved to: {local_image_path}")
|
||
return local_image_path
|
||
raise RuntimeError("screenshot file not created")
|
||
except Exception as e:
|
||
last_error = e
|
||
self.logger.warning(f"Failed to take screenshot (attempt {attempt + 1}/2): {e}")
|
||
time.sleep(0.2)
|
||
|
||
self.logger.error(f"Failed to take screenshot: {last_error}")
|
||
return None
|
||
|
||
def unlock(self) -> None:
|
||
"""Web平台无需解锁屏幕，实现空方法"""
|
||
pass
|
||
|
||
def get_current_url(self) -> str:
|
||
"""
|
||
获取当前页面URL
|
||
|
||
:return: 当前页面URL
|
||
"""
|
||
if self.driver:
|
||
try:
|
||
if self.engine == "playwright":
|
||
self._current_url = self.driver.url
|
||
else:
|
||
self._current_url = self.driver.current_url
|
||
return self._current_url
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to get current URL: {e}")
|
||
return self._current_url
|
||
|
||
def get_page_title(self) -> str:
|
||
"""
|
||
获取当前页面标题
|
||
|
||
:return: 当前页面标题
|
||
"""
|
||
if self.driver:
|
||
try:
|
||
if self.engine == "playwright":
|
||
return self.driver.title()
|
||
else:
|
||
return self.driver.title
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to get page title: {e}")
|
||
return ""
|
||
|
||
# ==================== 事件发送 ====================
|
||
|
||
def send_event(self, event) -> bool:
|
||
"""发送Web事件"""
|
||
try:
|
||
self.logger.debug(f"Sending Web event: {event}")
|
||
success = event.send(self)
|
||
if success and self.engine != "playwright":
|
||
self._sync_window_handles()
|
||
return bool(success)
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to send Web event: {e}")
|
||
return False
|
||
|
||
# ==================== 应用管理 ====================
|
||
|
||
@property
|
||
def app_identifier(self) -> str:
|
||
"""获取Web应用的唯一标识符（URL）"""
|
||
if self._app:
|
||
return self._app.identifier
|
||
return ""
|
||
|
||
def _get_domain(self, url: str) -> str:
|
||
"""从 URL 中提取二级域名"""
|
||
if not url:
|
||
return ""
|
||
from urllib.parse import urlparse
|
||
try:
|
||
netloc = urlparse(url).netloc
|
||
if not netloc:
|
||
return ""
|
||
parts = netloc.split('.')
|
||
if len(parts) >= 2:
|
||
# 取最后两部分，例如 www.baidu.com -> baidu.com
|
||
return ".".join(parts[-2:])
|
||
return netloc
|
||
except Exception:
|
||
return ""
|
||
|
||
def is_foreground(self) -> bool:
|
||
"""检查Web应用是否在前台（基于域名匹配）"""
|
||
if not self.driver:
|
||
return False
|
||
|
||
try:
|
||
current_url = self.get_current_url()
|
||
# 检查当前URL是否与目标应用二级域名匹配
|
||
if self._app:
|
||
target_domain = self._get_domain(self._app.identifier)
|
||
current_domain = self._get_domain(current_url)
|
||
return target_domain == current_domain
|
||
return False
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to check if Web app is in foreground: {e}")
|
||
return False
|
||
|
||
def get_redirect_target_info(self):
|
||
"""
|
||
Web平台跳转检测：基于域名变化判断是否离开目标站点
|
||
|
||
- 当前域名匹配目标 → 返回 None（在前台）
|
||
- 当前域名不匹配 → 返回 {"target": domain, "type": "other"}
|
||
- 无法获取URL → 返回 {"target": None, "type": "unknown"}
|
||
|
||
:return: 跳转信息字典或 None
|
||
"""
|
||
if not self.driver:
|
||
return {"target": None, "type": "unknown"}
|
||
try:
|
||
current_url = self.get_current_url()
|
||
if self._app:
|
||
target_domain = self._get_domain(self._app.identifier)
|
||
current_domain = self._get_domain(current_url)
|
||
if target_domain == current_domain:
|
||
return None # 仍在目标站点
|
||
return {"target": current_domain, "type": "other"}
|
||
return None
|
||
except Exception:
|
||
return {"target": None, "type": "unknown"}
|
||
|
||
def check_network(self, host: str = "8.8.8.8") -> bool:
|
||
"""
|
||
通过浏览器内 JS 检测网络连通性
|
||
|
||
优先使用 XMLHttpRequest 同步请求 Google generate_204，
|
||
失败时 fallback 到 navigator.onLine。
|
||
|
||
:param host: 未使用（保持接口兼容）
|
||
:return: 网络是否可用
|
||
"""
|
||
if not self.driver:
|
||
return False
|
||
try:
|
||
script = """
|
||
try {
|
||
var xhr = new XMLHttpRequest();
|
||
xhr.open('HEAD', 'https://www.google.com/generate_204', false);
|
||
xhr.timeout = 5000;
|
||
xhr.send();
|
||
return xhr.status === 204 || xhr.status === 200;
|
||
} catch(e) {
|
||
return navigator.onLine;
|
||
}
|
||
"""
|
||
if self.engine == "playwright":
|
||
result = self.driver.evaluate(f"() => {{ {script} }}")
|
||
else:
|
||
result = self.driver.execute_script(script)
|
||
return bool(result)
|
||
except Exception as e:
|
||
self.logger.warning(f"check_network 异常: {e}")
|
||
return False
|
||
|
||
def pull_back_to_app(self) -> bool:
|
||
"""将Web应用拉回前台"""
|
||
app_url = self.app_url
|
||
if not self.driver or not app_url:
|
||
return False
|
||
|
||
try:
|
||
if self.engine == "playwright":
|
||
self.driver.goto(app_url)
|
||
self._wait_for_page_ready()
|
||
else:
|
||
self.driver.get(app_url)
|
||
self._wait_for_page_ready()
|
||
self.logger.info(f"Pulled back to Web app: {app_url}")
|
||
return True
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to pull back Web app to foreground: {e}")
|
||
return False
|
||
|
||
def start_app(self) -> bool:
|
||
"""启动Web应用"""
|
||
app_url = self.app_url
|
||
if not app_url:
|
||
self.logger.warning("No Web app URL specified, cannot start")
|
||
return False
|
||
|
||
try:
|
||
if not self.driver:
|
||
self.set_up()
|
||
|
||
if self.engine == "playwright":
|
||
self.driver.goto(app_url)
|
||
self._wait_for_page_ready()
|
||
self._current_url = self.driver.url
|
||
else:
|
||
self.driver.get(app_url)
|
||
self._wait_for_page_ready()
|
||
self._current_url = self.driver.current_url
|
||
self.logger.info(f"Started Web app: {self._current_url}")
|
||
return True
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to start Web app: {e}")
|
||
return False
|
||
|
||
# ==================== Web 特定方法 ====================
|
||
|
||
def _sync_window_handles(self) -> None:
|
||
"""
|
||
P2: 带时间戳缓存的窗口句柄同步
|
||
|
||
500ms内不重复调用window_handles，减少IPC开销。
|
||
"""
|
||
if not self.driver or getattr(self, "engine", "selenium") == "playwright":
|
||
return
|
||
|
||
import time
|
||
now = time.time()
|
||
if now - self._last_sync_time < 0.5:
|
||
return # 距上次同步不到500ms，跳过
|
||
self._last_sync_time = now
|
||
|
||
try:
|
||
current_handles = self.driver.window_handles
|
||
# 检查是否有新窗口
|
||
new_handles = [h for h in current_handles if h not in self._window_handles]
|
||
|
||
if new_handles:
|
||
# 切换到最新的窗口
|
||
target_window = new_handles[-1]
|
||
self.driver.switch_to.window(target_window)
|
||
self._window_handles.update(current_handles)
|
||
self.logger.info(f"Detected new window, switched to: {target_window}")
|
||
else:
|
||
# 检查当前窗口是否仍然有效，如果无效则切换回一个有效的窗口
|
||
try:
|
||
_ = self.driver.current_window_handle
|
||
except Exception:
|
||
self.logger.warning("Current window closed, switching to last valid handle")
|
||
if current_handles:
|
||
self.driver.switch_to.window(current_handles[-1])
|
||
self._window_handles = set(current_handles)
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to sync window handles: {e}")
|
||
|
||
def refresh_page(self) -> None:
|
||
"""刷新当前页面"""
|
||
if self.driver:
|
||
try:
|
||
if getattr(self, "engine", "selenium") == "playwright":
|
||
self.driver.reload()
|
||
else:
|
||
self.driver.refresh()
|
||
except Exception as e:
|
||
self.logger.error(f"Failed to refresh page: {e}")
|