# Selenium Chrome 浏览器适配。合同和返利共用这一份。 # # 通过 WebDriver 操控 Chrome。用户目录必须在仓库外，避免 list_dir 扫到数 GB 的 profile。 # 端口探测和 DevToolsActivePort 用短超时轮询，禁止无条件 sleep(1)。 # 返利用户 Chrome 的 junction / 端口由 rpa_rebate.apa_browser_adapter 覆盖。 from __future__ import annotations import os import socket import subprocess import time from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] _FLOW_DEBUG_PORTS = { "CONTRACT_CREATE": 9333, "REBATE_FIND": 9335, "default": 9334, } _IGNORE_SOURCE_JS = """ function vis(el) { return !!(el && (el.offsetWidth || el.offsetHeight)); } function compact(el) { return ((el && (el.innerText || '')) || '').replace(/\\s+/g, ' ').trim(); } const hits = Array.from(document.querySelectorAll('.vxe-modal--wrapper')).filter((el) => { if (!vis(el)) return false; const title = compact(el.querySelector('.vxe-modal--header-title, .vxe-modal--title')); return title === '货源选择'; }); for (const modal of hits) { const btn = Array.from(modal.querySelectorAll('button')).find((el) => { const t = compact(el).replace(/ /g, ''); return vis(el) && (t === '关闭' || t === '取消'); }); if (btn) { btn.click(); continue; } const x = modal.querySelector('.vxe-modal--close-btn, .vxe-icon-close'); if (x) { x.click(); } } return hits.length; """ # 判断 path 是否落在仓库目录树内。 def _is_inside_repo(path: Path, repo_root: Path) -> bool: try: path.resolve().relative_to(repo_root.resolve()) except ValueError: return False return True # 把 flow 名收成目录名安全的一段。空则 default。 def _sanitize_flow(flow: str) -> str: safe = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in (flow or "default").strip()) return safe or "default" # 解析 Chrome user-data 根目录，必须留在仓库外。 # # 未指定时用仓库父目录下的 rpa-chrome-profile。RPA_CHROME_PROFILE 指到仓库内则改回这个默认。 def resolve_chrome_profile_base(requested: str | None = None, *, repo_root: Path | None = None) -> Path: root = (repo_root or REPO_ROOT).resolve() outside = (root.parent / "rpa-chrome-profile").resolve() raw = os.getenv("RPA_CHROME_PROFILE") if requested is None else requested if raw: candidate = Path(raw).expanduser().resolve() if _is_inside_repo(candidate, root): print(f"RPA_CHROME_PROFILE 指向仓库内 {candidate}，已改用 {outside}", flush=True) return outside return candidate return outside # 按 flow 给出稳定的 Chrome 用户目录，后续班次复用登录态。 def resolve_chrome_profile_dir( flow: str = "default", requested: str | None = None, *, repo_root: Path | None = None, ) -> Path: base = resolve_chrome_profile_base(requested, repo_root=repo_root) return (base / _sanitize_flow(flow)).resolve() # 给出本 flow 的调试口。RPA_CHROME_DEBUG_PORT 优先；否则按班次表，未知回 9333。 def chrome_debug_port(flow: str = "default") -> int: raw = (os.getenv("RPA_CHROME_DEBUG_PORT") or "").strip() if raw: try: return int(raw) except ValueError: pass return _FLOW_DEBUG_PORTS.get(_sanitize_flow(flow), 9333) # 探测 127.0.0.1:port 此刻能否连上。 def _port_open(port: int) -> bool: try: with socket.create_connection(("127.0.0.1", port), timeout=0.4): return True except OSError: return False # 从 profile 的 DevToolsActivePort 读出当前调试口，没有或坏文件返回 None。 def _read_devtools_port(profile_dir: Path) -> int | None: marker = profile_dir / "DevToolsActivePort" if not marker.is_file(): return None try: first = marker.read_text(encoding="utf-8").splitlines()[0].strip() port = int(first) except (OSError, ValueError, IndexError): return None return port if port > 0 else None # 轮询本机调试口直到可连上或超时。间隔 0.3 秒，不 sleep(1)。 def wait_debug_port(port: int, timeout: float = 20) -> bool: deadline = time.time() + timeout while time.time() <= deadline: if _port_open(port): return True time.sleep(0.3) return False # 轮询 DevToolsActivePort 直到读到正端口或超时。间隔 0.3 秒。 def wait_devtools_port(profile_dir: Path, timeout: float = 20) -> int | None: deadline = time.time() + timeout while time.time() <= deadline: port = _read_devtools_port(profile_dir) if port: return port time.sleep(0.3) return None # 构造 ChromeDriver Service。有 CHROMEDRIVER_PATH 用本地，否则 webdriver_manager 下载。 def _chrome_service(): from selenium.webdriver.chrome.service import Service chromedriver = os.getenv("CHROMEDRIVER_PATH") if chromedriver and os.path.exists(chromedriver): return Service(chromedriver) from webdriver_manager.chrome import ChromeDriverManager return Service(ChromeDriverManager().install()) # 找到本机 chrome 可执行文件。CHROME_PATH 优先，找不到则失败。 def _chrome_binary() -> str: raw = (os.getenv("CHROME_PATH") or "").strip() local = Path(os.environ.get("LOCALAPPDATA", "")) / "Google" / "Chrome" / "Application" / "chrome.exe" candidates = [ raw, r"C:\Program Files\Google\Chrome\Application\chrome.exe", r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe", str(local), "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", ] for path in candidates: if path and os.path.isfile(path): return path raise RuntimeError("未找到 chrome.exe") # 附着到已在调试口上的 Chrome，并跑一句脚本确认活着。 def _attach_chrome(service, port: int): from selenium import webdriver from selenium.webdriver.chrome.options import Options options = Options() options.add_experimental_option("debuggerAddress", f"127.0.0.1:{port}") driver = webdriver.Chrome(service=service, options=options) driver.execute_script("return 1") return driver # 独立进程拉起 Chrome，不跟 Python / chromedriver 一起退出。 def _spawn_chrome(profile_dir: Path, port: int, *, restore_session: bool = False) -> None: chrome = _chrome_binary() cmd = [ chrome, f"--remote-debugging-port={str(port)}", "--user-data-dir=" + str(profile_dir), "--remote-allow-origins=*", "--start-maximized", "--no-first-run", "--no-default-browser-check", "--disable-popup-blocking", ] if restore_session: cmd.append("--restore-last-session") kwargs: dict = { "stdin": subprocess.DEVNULL, "stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL, } if os.name == "nt": kwargs["close_fds"] = False kwargs["creationflags"] = 0x00000008 | 0x00000200 | 0x01000000 subprocess.Popen(cmd, **kwargs) # 强制结束 chrome.exe，给用户目录重开调试口腾位置。仅 Windows。 def _kill_chrome() -> None: if os.name != "nt": return creation = getattr(subprocess, "CREATE_NO_WINDOW", 0) subprocess.run( ["taskkill", "/IM", "chrome.exe", "/F"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, creationflags=creation, ) # 等到 profile 的 SingletonLock / lockfile 消失，或超时放弃。 def _wait_profile_unlocked(profile_dir: Path, timeout: float = 20) -> None: deadline = time.time() + timeout while time.time() <= deadline: locked = (profile_dir / "SingletonLock").exists() or (profile_dir / "lockfile").exists() if not locked: return time.sleep(0.3) # 独立拉起 Chrome 再附着调试口。端口和 DevTools 用轮询，不 sleep(1)。 # # 口已开则直接附着。restore_session 时先结束旧进程、等 profile 解锁再拉起。 def launch_independent_chrome(service, profile_dir: Path, port: int, *, restore_session: bool = False): if not _port_open(port): if restore_session: print( f"Chrome 日常窗口无调试口，按同一用户目录重开 port={port} profile={profile_dir}", flush=True, ) _kill_chrome() _wait_profile_unlocked(profile_dir) _spawn_chrome(profile_dir, port, restore_session=restore_session) if not wait_debug_port(port): raise RuntimeError(f"Chrome 未在端口 {port} 起来") wait_devtools_port(profile_dir, timeout=5) return _attach_chrome(service, port) # 用 chromedriver 新开带调试口的 Chrome，起来后再轮询调试口。 def _launch_chrome_selenium(service, profile_dir: Path, port: int): from selenium import webdriver from selenium.webdriver.chrome.options import Options options = Options() options.page_load_strategy = "eager" options.add_experimental_option("detach", True) options.add_experimental_option("excludeSwitches", ["enable-automation", "enable-logging"]) 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") options.add_argument("--disable-gpu") options.add_argument("--disable-extensions") options.add_argument("--disable-popup-blocking") options.add_argument("--start-maximized") options.add_argument("--remote-allow-origins=*") options.add_argument(f"--remote-debugging-port={port}") options.add_argument("--user-data-dir=" + str(profile_dir)) driver = webdriver.Chrome(service=service, options=options) wait_debug_port(port, timeout=5) return driver # Selenium Chrome 浏览器适配器，实现合同 / 返利页面流用的 BrowserAdapter 协议。 # # 子类可覆盖 profile / 调试口 / 拉起方式。返利用户 Chrome 走这条扩展点。 class ApaBrowserAdapter: # 记下起始 URL 和班次名。真正开浏览器推迟到第一次页面操作。 def __init__(self, url: str = "", *, flow: str = "default"): self._url = url self._flow = flow self._driver = None self._is_open = False self._download_dir = "" self._download_before: set[str] = set() self._ignoring_source = False self._hold_source_choice = False # 本班次 Chrome 用户目录。默认按 flow 分子目录。 def _profile_dir(self) -> Path: return resolve_chrome_profile_dir(self._flow) # 本班次调试口。默认按 flow 查表。 def _debug_port(self) -> int: return chrome_debug_port(self._flow) # 是否按日常 Chrome 用户目录重开并恢复标签。默认否。 def _restore_session(self) -> bool: return False # 若已有目标页签则切过去。默认不切。 def _focus_existing_tab(self, driver) -> bool: return False # 新开 Chrome。默认 chromedriver；返利子类改成独立进程。 def _launch_driver(self, service, profile_dir: Path, port: int): return _launch_chrome_selenium(service, profile_dir, port) # 复用本班次已打开的 Chrome；没有就按稳定 profile 新开。 def _get_driver(self): if self._driver is not None: return self._driver profile_dir = self._profile_dir() if _is_inside_repo(profile_dir, REPO_ROOT): raise RuntimeError(f"拒绝在仓库内创建 Chrome 用户目录: {profile_dir}") os.makedirs(profile_dir, exist_ok=True) port = self._debug_port() if not self._restore_session(): port = _read_devtools_port(profile_dir) or port driver = None if _port_open(port): try: driver = _attach_chrome(_chrome_service(), port) print(f"Chrome 复用已有实例 flow={self._flow} port={port}", flush=True) except Exception as exc: print(f"Chrome 复用失败 port={port}：{exc}", flush=True) driver = None if driver is None: launch_port = self._debug_port() try: driver = self._launch_driver(_chrome_service(), profile_dir, launch_port) print( f"Chrome 新开实例 flow={self._flow} port={launch_port} profile={profile_dir}", flush=True, ) except Exception as exc: retry_port = wait_devtools_port(profile_dir, timeout=2) or _read_devtools_port(profile_dir) if retry_port and _port_open(retry_port): driver = _attach_chrome(_chrome_service(), retry_port) print(f"Chrome 锁定后改为复用 port={retry_port}", flush=True) else: raise RuntimeError(f"打开 Chrome 失败：{exc}") from exc driver.set_page_load_timeout(45) driver.set_script_timeout(30) self._driver = driver return self._driver # 保证浏览器已开。空白页才打开起始 URL；子类可先切到已有页签。 def _ensure_open(self): if self._driver is None: driver = self._get_driver() href = "" try: href = str(driver.execute_script("return location.href") or "") except Exception: href = "" blank = (not href) or href.startswith("chrome://") or href.startswith("about:") if self._focus_existing_tab(driver): print("Chrome 切到已打开的采购协议页", flush=True) elif self._url and blank: driver.get(self._url) self._is_open = True # 按 CSS / XPath / text= / label= / aria= / title= / :contains() 找第一个元素。 # # timeout 保留给调用方语义，底层 find_element 本身不等待。 def _find(self, selector: str, timeout: int = 5): from selenium.webdriver.common.by import By from selenium.common.exceptions import NoSuchElementException import re if selector.startswith("text="): text_val = selector[5:].strip('"').strip("'") xpath = f'//*[contains(text(),"{text_val}")]' return self._driver.find_element(By.XPATH, xpath) if selector.startswith("label="): text_val = selector[6:].strip('"').strip("'") xpath = ( f'//label[contains(text(),"{text_val}")]' f"/following-sibling::*[1][self::input or self::select or self::textarea]" ) try: return self._driver.find_element(By.XPATH, xpath) except NoSuchElementException: 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 if selector.startswith("aria="): text_val = selector[5:].strip('"').strip("'") return self._driver.find_element(By.CSS_SELECTOR, f'[aria-label="{text_val}"]') if selector.startswith("title="): text_val = selector[6:].strip('"').strip("'") return self._driver.find_element(By.CSS_SELECTOR, f'[title="{text_val}"]') 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) if selector.startswith("//") or selector.startswith("/"): return self._driver.find_element(By.XPATH, selector) return self._driver.find_element(By.CSS_SELECTOR, selector) # 打开指定 URL。eager 超时但文档已出来时继续，不把长轮询当失败。 def open_url(self, url: str) -> None: self._url = url driver = self._get_driver() try: driver.get(url) except Exception as exc: current = "" try: current = driver.current_url or "" except Exception: current = "" if not current or current in ("data:,", "about:blank"): raise RuntimeError(f"打开页面失败：{url} ({exc})") from exc self._is_open = True # 用原生 value setter 加 input/change/blur 注入，兼容 LUI/React 受控组件。 def _set_value_via_js(self, el, value: str) -> None: 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), ) # 按选择器列表填第一个能写上的输入框。原生失败则改 JS 注入。 def fill_first(self, selectors: list, value: str) -> None: self._ensure_open() last_error = None for sel in selectors: try: el = self._find(sel) 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) self._ignore_source_bug() return except Exception as e: last_error = e continue raise RuntimeError(f"fill_first failed: {selectors}, last: {last_error}") # 真实鼠标点击，给只监听 mousedown 的 LUI 页签/工具栏用。 def click_native(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) self._driver.execute_script( "arguments[0].scrollIntoView({block:'center', inline:'center'});", el, ) time.sleep(0.2) ActionChains(self._driver).move_to_element(el).pause(0.15).click().perform() time.sleep(0.4) self._ignore_source_bug() return except Exception as e: last_error = e continue raise RuntimeError(f"click_native failed: {selectors}, last: {last_error}") # 按选择器列表 JS 点击第一个能点到的元素。 def click_first(self, selectors: list) -> None: self._ensure_open() last_error = None for sel in selectors: try: el = self._find(sel) 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}") # 等到页面 body 出现指定文本，超时抛 TimeoutError。 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}") # 等到任一选择器能找到元素，超时抛 TimeoutError。 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}") # 任一选择器此刻能找到元素则 True。 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 # 切换到匹配到的 iframe。优先可见弹窗，避免切到上次留下的隐藏 iframe。 def switch_to_frame(self, selectors: list) -> None: self._ensure_open() from selenium.webdriver.common.by import By last_error = None found = [] for sel in selectors: try: if sel.startswith("//") or sel.startswith("/"): found.extend(self._driver.find_elements(By.XPATH, sel)) else: found.extend(self._driver.find_elements(By.CSS_SELECTOR, sel)) except Exception as e: last_error = e visible = [] hidden = [] for el in found: try: (visible if el.is_displayed() else hidden).append(el) except Exception: hidden.append(el) for el in visible + hidden: try: self._driver.switch_to.default_content() self._driver.switch_to.frame(el) return except Exception as e: last_error = e raise RuntimeError(f"switch_to_frame failed: {selectors}, last: {last_error}") # 切回主文档（顶层 frame）。 def switch_to_default_content(self) -> None: self._ensure_open() self._driver.switch_to.default_content() # 看见「货源选择」bug 弹层就关，不参与任何选择。 def _ignore_source_bug(self) -> None: if self._hold_source_choice or self._ignoring_source or self._driver is None: return self._ignoring_source = True try: n = self._driver.execute_script(_IGNORE_SOURCE_JS) if n: time.sleep(0.2) except Exception: pass finally: self._ignoring_source = False # 在当前页执行一段 JS 并返回结果。 def execute_script(self, script: str, *args): self._ensure_open() result = self._driver.execute_script(script, *args) self._ignore_source_bug() return result # 选中 Landray xform fancy-select（.fs-wrap / .fs-option）。 def select_xform(self, name: str, value: str) -> None: self._ensure_open() result = self._driver.execute_script( """ const name = arguments[0]; const wanted = String(arguments[1] || '').trim(); const flag = document.querySelector('xformflag[property="' + name + '"]'); if (!flag) { return {ok: false, error: 'flag not found: ' + name}; } const norm = (s) => String(s || '').replace(/\\s+/g, '').trim(); const wantedN = norm(wanted); const opts = Array.from(flag.querySelectorAll('.fs-option')); let target = opts.find((o) => { const text = norm((o.querySelector('.fs-option-label') || o).textContent); const val = norm(o.getAttribute('data-value')); return text === wantedN || val === wantedN; }); if (!target && wantedN.length >= 2) { target = opts.find((o) => { const text = norm((o.querySelector('.fs-option-label') || o).textContent); return text.indexOf(wantedN) >= 0; }); } if (!target) { return { ok: false, error: 'option not found: ' + wanted, available: opts.map((o) => ((o.querySelector('.fs-option-label') || o).textContent || '').trim()), }; } const labelWrap = flag.querySelector('.fs-label-wrap'); if (labelWrap) { labelWrap.click(); } target.click(); const input = flag.querySelector('input[name="' + name + '"]'); const selected = target.getAttribute('data-value') || ''; if (input) { const proto = window.HTMLInputElement.prototype; const setter = Object.getOwnPropertyDescriptor(proto, 'value').set; setter.call(input, selected); input.dispatchEvent(new Event('input', { bubbles: true })); input.dispatchEvent(new Event('change', { bubbles: true })); if (typeof __xformDispatch === 'function') { try { __xformDispatch(selected, input); } catch (e) {} } } return { ok: true, value: selected, label: ((flag.querySelector('.fs-label') || {}).textContent || '').trim(), }; """, name, str(value), ) if not result or not result.get("ok"): raise RuntimeError(f"select_xform failed: {name}={value} {result}") time.sleep(0.3) # 按选择器列表选中原生 select 的可见文本，失败再用 JS 点 option。 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) try: Select(el).select_by_visible_text(value) time.sleep(0.3) if self._selected_text_matches(el, value): return except Exception: pass 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}") # 当前选中项的可见文本是否等于 value。 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 # 用 JS 按可见文本或 value 选中 option 并派发 change。 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), ) # 按选择器列表给第一个 file 输入框送本地路径。 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}") # 把浏览器下载目录指到 path，并记下当前已有文件名以便 wait_for_download 认新文件。 def set_download_dir(self, path: str) -> None: self._ensure_open() dest = str(Path(path).expanduser().resolve()) Path(dest).mkdir(parents=True, exist_ok=True) self._download_dir = dest self._download_before = {p.name for p in Path(dest).iterdir()} try: self._driver.execute_cdp_cmd( "Page.setDownloadBehavior", {"behavior": "allow", "downloadPath": dest, "eventsEnabled": True}, ) except Exception: try: self._driver.execute_cdp_cmd( "Browser.setDownloadBehavior", {"behavior": "allow", "downloadPath": dest, "eventsEnabled": True}, ) except Exception as exc: print(f"set_download_dir skipped: {exc}", flush=True) # 等到下载目录出现新的 suffix 文件（忽略 .crdownload），返回最新那条路径。 def wait_for_download(self, suffix: str = ".xlsx", timeout_seconds: int = 20) -> str: dest = Path(self._download_dir or "") if not dest.is_dir(): raise RuntimeError("未设置下载目录") before = set(self._download_before) deadline = time.time() + timeout_seconds wanted = (suffix or ".xlsx").lower() while time.time() <= deadline: hits = [ p for p in dest.iterdir() if p.name not in before and p.suffix.lower() == wanted and not str(p).endswith(".crdownload") and not Path(str(p) + ".crdownload").exists() ] if hits: return str(max(hits, key=lambda p: p.stat().st_mtime)) time.sleep(0.3) raise RuntimeError(f"未下到 {wanted} 文件") # 按选择器列表读第一个非空 text/value。 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}") # 读当前页 body 可见文本，读不到返回空串。 def read_body_text(self) -> str: self._ensure_open() try: return self._driver.find_element("tag name", "body").text except Exception: return "" # 把当前页截到本包 accept.artifact_dir(flow) 下的 name.png，返回路径。 def screenshot(self, name: str) -> str: from accept import artifact_dir save_dir = artifact_dir(self._flow) save_dir.mkdir(parents=True, exist_ok=True) path = save_dir / f"{name}.png" if self._driver: self._driver.save_screenshot(str(path)) return str(path) # 接受原生 alert/confirm 并返回其文本。没有弹窗则空串。 def accept_alert(self) -> str: self._ensure_open() try: alert = self._driver.switch_to.alert text = (alert.text or "").strip() alert.accept() return text except Exception: return "" # 关掉当前标签，在同一 Chrome 里新开一标签再打开 url。 def reopen_url(self, url: str) -> None: driver = self._get_driver() old = None try: old = driver.current_window_handle driver.switch_to.new_window("tab") new = driver.current_window_handle if old and old != new: try: driver.switch_to.window(old) driver.close() except Exception: pass try: driver.switch_to.window(new) except Exception: handles = driver.window_handles if handles: driver.switch_to.window(handles[-1]) except Exception: pass self.open_url(url)