344 lines
13 KiB
Python
344 lines
13 KiB
Python
# coding=utf-8
|
||
"""
|
||
APA (艺赛旗) Selenium Chrome BrowserAdapter 实现
|
||
通过 Selenium WebDriver 直接操控 Chrome，不依赖 APA 浏览器模块
|
||
"""
|
||
|
||
import time
|
||
|
||
|
||
class ApaBrowserAdapter:
|
||
"""Selenium Chrome 浏览器适配器，实现 BrowserAdapter 协议"""
|
||
|
||
def __init__(self, url: str = ""):
|
||
self._url = url
|
||
self._driver = None
|
||
self._is_open = False
|
||
|
||
def _get_driver(self):
|
||
"""创建或获取 Chrome WebDriver"""
|
||
if self._driver is not None:
|
||
return self._driver
|
||
|
||
from selenium import webdriver
|
||
from selenium.webdriver.chrome.service import Service
|
||
from selenium.webdriver.chrome.options import Options
|
||
from webdriver_manager.chrome import ChromeDriverManager
|
||
|
||
options = Options()
|
||
# 不自动关闭浏览器，方便调试
|
||
options.add_experimental_option("detach", True)
|
||
# 避免被检测为自动化
|
||
options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
||
options.add_experimental_option("useAutomationExtension", False)
|
||
# 基本参数
|
||
options.add_argument("--disable-blink-features=AutomationControlled")
|
||
options.add_argument("--no-first-run")
|
||
options.add_argument("--no-default-browser-check")
|
||
|
||
service = Service(ChromeDriverManager().install())
|
||
self._driver = webdriver.Chrome(service=service, options=options)
|
||
return self._driver
|
||
|
||
def _ensure_open(self):
|
||
if self._driver is None:
|
||
driver = self._get_driver()
|
||
if self._url:
|
||
driver.get(self._url)
|
||
self._is_open = True
|
||
|
||
# ---- 内部查找方法 ----
|
||
|
||
def _find(self, selector: str, timeout: int = 5):
|
||
"""
|
||
智能查找元素：支持 CSS、XPath、text=、label=、aria=、title=、:contains()
|
||
"""
|
||
from selenium.webdriver.common.by import By
|
||
from selenium.common.exceptions import NoSuchElementException
|
||
import re
|
||
|
||
# 1. text=xxx → 查找包含对应文本的元素
|
||
if selector.startswith("text="):
|
||
text_val = selector[5:].strip('"').strip("'")
|
||
xpath = f'//*[contains(text(),"{text_val}")]'
|
||
return self._driver.find_element(By.XPATH, xpath)
|
||
|
||
# 2. label=xxx → 找到对应标签的输入框
|
||
if selector.startswith("label="):
|
||
text_val = selector[6:].strip('"').strip("'")
|
||
xpath = f'//label[contains(text(),"{text_val}")]/following-sibling::*[1][self::input or self::select or self::textarea]'
|
||
try:
|
||
return self._driver.find_element(By.XPATH, xpath)
|
||
except NoSuchElementException:
|
||
# Fallback: find the label's "for" attribute
|
||
try:
|
||
for_id = self._driver.find_element(
|
||
By.XPATH, f'//label[contains(text(),"{text_val}")]'
|
||
).get_attribute("for")
|
||
if for_id:
|
||
return self._driver.find_element(By.ID, for_id)
|
||
except NoSuchElementException:
|
||
pass
|
||
raise
|
||
|
||
# 3. aria=xxx → aria-label 属性
|
||
if selector.startswith("aria="):
|
||
text_val = selector[5:].strip('"').strip("'")
|
||
return self._driver.find_element(By.CSS_SELECTOR, f'[aria-label="{text_val}"]')
|
||
|
||
# 4. title=xxx → title 属性
|
||
if selector.startswith("title="):
|
||
text_val = selector[6:].strip('"').strip("'")
|
||
return self._driver.find_element(By.CSS_SELECTOR, f'[title="{text_val}"]')
|
||
|
||
# 5. :contains() → 转 XPath（支持 button:contains(文本) / a:contains("文本")）
|
||
if ":contains(" in selector:
|
||
m = re.match(r'(.+?):contains\(["\']?(.+?)["\']?\)', selector)
|
||
if m:
|
||
tag_part, text_val = m.groups()
|
||
xpath = f'//{tag_part}[contains(.,"{text_val}")]'
|
||
return self._driver.find_element(By.XPATH, xpath)
|
||
|
||
# 6. 显式 XPath（以 / 或 // 开头）优先按 XPath 查找
|
||
if selector.startswith("//") or selector.startswith("/"):
|
||
return self._driver.find_element(By.XPATH, selector)
|
||
|
||
# 7. 标准 CSS selector
|
||
return self._driver.find_element(By.CSS_SELECTOR, selector)
|
||
|
||
# ---- BrowserAdapter 协议实现 ----
|
||
|
||
def open_url(self, url: str) -> None:
|
||
self._url = url
|
||
self._get_driver().get(url)
|
||
self._is_open = True
|
||
|
||
def _set_value_via_js(self, el, value: str) -> None:
|
||
"""通过原生 value setter + 事件注入设置值，兼容 LUI/React 受控组件"""
|
||
self._driver.execute_script(
|
||
"""
|
||
var el = arguments[0];
|
||
var value = arguments[1];
|
||
var proto = el.tagName === 'TEXTAREA'
|
||
? window.HTMLTextAreaElement.prototype
|
||
: window.HTMLInputElement.prototype;
|
||
var setter = Object.getOwnPropertyDescriptor(proto, 'value').set;
|
||
setter.call(el, value);
|
||
el.dispatchEvent(new Event('input', { bubbles: true }));
|
||
el.dispatchEvent(new Event('change', { bubbles: true }));
|
||
el.dispatchEvent(new Event('blur', { bubbles: true }));
|
||
""",
|
||
el,
|
||
str(value),
|
||
)
|
||
|
||
def fill_first(self, selectors: list, value: str) -> None:
|
||
self._ensure_open()
|
||
last_error = None
|
||
for sel in selectors:
|
||
try:
|
||
el = self._find(sel)
|
||
# 先尝试原生输入；LUI 受控组件用 JS 注入兜底
|
||
try:
|
||
el.clear()
|
||
el.send_keys(str(value))
|
||
except Exception:
|
||
pass
|
||
if el.get_attribute("value") != str(value):
|
||
self._set_value_via_js(el, str(value))
|
||
time.sleep(0.3)
|
||
return
|
||
except Exception as e:
|
||
last_error = e
|
||
continue
|
||
raise RuntimeError(f"fill_first failed: {selectors}, last: {last_error}")
|
||
|
||
def click_first(self, selectors: list) -> None:
|
||
self._ensure_open()
|
||
last_error = None
|
||
for sel in selectors:
|
||
try:
|
||
el = self._find(sel)
|
||
# LUI/委托事件用 JS 点击最可靠
|
||
self._driver.execute_script("arguments[0].click()", el)
|
||
time.sleep(0.5)
|
||
return
|
||
except Exception as e:
|
||
last_error = e
|
||
continue
|
||
raise RuntimeError(f"click_first failed: {selectors}, last: {last_error}")
|
||
|
||
def double_click_first(self, selectors: list) -> None:
|
||
self._ensure_open()
|
||
from selenium.webdriver.common.action_chains import ActionChains
|
||
last_error = None
|
||
for sel in selectors:
|
||
try:
|
||
el = self._find(sel)
|
||
ActionChains(self._driver).double_click(el).perform()
|
||
time.sleep(0.5)
|
||
return
|
||
except Exception as e:
|
||
last_error = e
|
||
continue
|
||
raise RuntimeError(f"double_click_first failed: {selectors}, last: {last_error}")
|
||
|
||
def wait_for_text(self, text: str, timeout_seconds: int = 15) -> None:
|
||
self._ensure_open()
|
||
from selenium.webdriver.support.ui import WebDriverWait
|
||
from selenium.webdriver.support import expected_conditions as EC
|
||
from selenium.webdriver.common.by import By
|
||
try:
|
||
WebDriverWait(self._driver, timeout_seconds).until(
|
||
EC.text_to_be_present_in_element((By.TAG_NAME, "body"), text)
|
||
)
|
||
except Exception as e:
|
||
raise TimeoutError(f"wait_for_text '{text}' timed out: {e}")
|
||
|
||
def wait_for_element(self, selectors: list, timeout_seconds: int = 15) -> None:
|
||
self._ensure_open()
|
||
deadline = time.time() + timeout_seconds
|
||
last_error = None
|
||
while time.time() < deadline:
|
||
for sel in selectors:
|
||
try:
|
||
self._find(sel, timeout=1)
|
||
return
|
||
except Exception as e:
|
||
last_error = e
|
||
time.sleep(0.5)
|
||
raise TimeoutError(f"wait_for_element timed out: {selectors}, last: {last_error}")
|
||
|
||
def element_exists(self, selectors: list) -> bool:
|
||
self._ensure_open()
|
||
for sel in selectors:
|
||
try:
|
||
self._find(sel, timeout=1)
|
||
return True
|
||
except Exception:
|
||
continue
|
||
return False
|
||
|
||
def switch_to_frame(self, selectors: list) -> None:
|
||
"""切换到匹配到的 iframe（客商选择弹窗等）。"""
|
||
self._ensure_open()
|
||
last_error = None
|
||
for sel in selectors:
|
||
try:
|
||
el = self._find(sel, timeout=1)
|
||
self._driver.switch_to.frame(el)
|
||
return
|
||
except Exception as e:
|
||
last_error = e
|
||
continue
|
||
raise RuntimeError(f"switch_to_frame failed: {selectors}, last: {last_error}")
|
||
|
||
def switch_to_default_content(self) -> None:
|
||
"""切回主文档（顶层 frame）。"""
|
||
self._ensure_open()
|
||
self._driver.switch_to.default_content()
|
||
|
||
def select_first(self, selectors: list, value: str) -> None:
|
||
self._ensure_open()
|
||
from selenium.webdriver.support.ui import Select
|
||
last_error = None
|
||
for sel in selectors:
|
||
try:
|
||
el = self._find(sel)
|
||
# 原生 <select>
|
||
try:
|
||
Select(el).select_by_visible_text(value)
|
||
time.sleep(0.3)
|
||
if self._selected_text_matches(el, value):
|
||
return
|
||
except Exception:
|
||
pass
|
||
# 兜底：JS 定位 option 并设置选中
|
||
self._select_via_js(el, value)
|
||
time.sleep(0.3)
|
||
if self._selected_text_matches(el, value):
|
||
return
|
||
raise RuntimeError(f"select_first 未选中目标项: {value}")
|
||
except Exception as e:
|
||
last_error = e
|
||
continue
|
||
raise RuntimeError(f"select_first failed: {selectors}, last: {last_error}")
|
||
|
||
def _selected_text_matches(self, el, value: str) -> bool:
|
||
try:
|
||
script = (
|
||
"var el = arguments[0];"
|
||
"if (el.selectedIndex != null) {"
|
||
" var o = el.options[el.selectedIndex];"
|
||
" return o ? (o.textContent || o.text || '').trim() : '';"
|
||
"}"
|
||
"return (el.value || '').trim();"
|
||
)
|
||
selected = self._driver.execute_script(script, el)
|
||
return (selected or "").strip() == str(value).strip()
|
||
except Exception:
|
||
return False
|
||
|
||
def _select_via_js(self, el, value: str) -> None:
|
||
self._driver.execute_script(
|
||
"""
|
||
var el = arguments[0];
|
||
var value = arguments[1];
|
||
var options = el.options || [];
|
||
for (var i = 0; i < options.length; i++) {
|
||
var t = (options[i].textContent || options[i].text || '').trim();
|
||
if (t === value || options[i].value === value) {
|
||
el.selectedIndex = i;
|
||
el.dispatchEvent(new Event('change', { bubbles: true }));
|
||
return;
|
||
}
|
||
}
|
||
""",
|
||
el,
|
||
str(value),
|
||
)
|
||
|
||
def upload_file_first(self, selectors: list, file_path: str) -> None:
|
||
self._ensure_open()
|
||
last_error = None
|
||
for sel in selectors:
|
||
try:
|
||
el = self._find(sel)
|
||
el.send_keys(file_path)
|
||
time.sleep(1)
|
||
return
|
||
except Exception as e:
|
||
last_error = e
|
||
continue
|
||
raise RuntimeError(f"upload_file_first failed: {selectors}, last: {last_error}")
|
||
|
||
def read_first(self, selectors: list, value_name: str = "") -> str:
|
||
self._ensure_open()
|
||
last_error = None
|
||
for sel in selectors:
|
||
try:
|
||
el = self._find(sel)
|
||
result = el.text or el.get_attribute("value") or ""
|
||
if result:
|
||
return result
|
||
except Exception as e:
|
||
last_error = e
|
||
continue
|
||
raise RuntimeError(f"read_first failed: {selectors}, last: {last_error}")
|
||
|
||
def read_body_text(self) -> str:
|
||
self._ensure_open()
|
||
try:
|
||
return self._driver.find_element("tag name", "body").text
|
||
except Exception:
|
||
return ""
|
||
|
||
def screenshot(self, name: str) -> str:
|
||
import os
|
||
save_dir = os.path.join(os.path.dirname(__file__), "screenshots")
|
||
os.makedirs(save_dir, exist_ok=True)
|
||
path = os.path.join(save_dir, f"{name}.png")
|
||
if self._driver:
|
||
self._driver.save_screenshot(path)
|
||
return path
|