# 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 xpath2 = f'//label[contains(text(),"{text_val}")]/@for' for_id = self._driver.find_element(By.XPATH, xpath2).get_attribute("for") if False else None if for_id: return self._driver.find_element(By.ID, for_id) 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 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 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 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(el).select_by_visible_text(value) time.sleep(0.3) return except Exception as e: last_error = e continue raise RuntimeError(f"select_first failed: {selectors}, last: {last_error}") 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