"""通用 Playwright 步骤执行器（不写死业务页面）。""" from __future__ import annotations import logging import re import time from datetime import datetime from pathlib import Path from typing import Any from browser_env import interpolate, resolve_url from browser_models import BrowserResult, BrowserScenario logger = logging.getLogger(__name__) SCREENSHOT_DIR = Path(__file__).resolve().parent / ".cache" / "screenshots" FIELD_HINTS: dict[str, list[str]] = { "账号": [ "#login-username", "input#login-username", "input[autocomplete='username']", "username", "account", "phone", "账号", "手机号", "企业账号", ], "密码": [ "#login-password input", "#login-password", "input#login-password", "input[type='password']", "password", "密码", ], "用户名": ["#login-username", "input#login-username", "username", "account", "账号"], } def _step_label(step: dict[str, Any], index: int) -> str: action = step.get("action", "?") target = step.get("target") or step.get("field") or step.get("url") or "" return f"{index + 1}. {action} {target}".strip() def _resolve_fill_locator(page, field: str, step: dict[str, Any]): if step.get("selector"): return page.locator(interpolate(str(step["selector"]))) field_key = interpolate(str(field)) if step.get("label"): return page.get_by_label(interpolate(str(step["label"])), exact=False) if step.get("placeholder"): return page.get_by_placeholder(interpolate(str(step["placeholder"])), exact=False) hints = FIELD_HINTS.get(field_key, [field_key]) for hint in hints: if hint.startswith("#") or hint.startswith(".") or hint.startswith("["): locator = page.locator(hint) if locator.count() > 0: return locator.first for getter in ( lambda h=hint: page.get_by_label(h, exact=False), lambda h=hint: page.get_by_placeholder(h, exact=False), ): locator = getter() if locator.count() > 0: return locator.first return page.locator("input, textarea").filter(has_text=field_key).first def _fill_field(page, field: str, step: dict[str, Any]) -> None: value = interpolate(str(step.get("value", ""))) locator = _resolve_fill_locator(page, field, step) locator.click(timeout=10_000) locator.fill("", timeout=5_000) locator.fill(value, timeout=10_000) def _page_error_text(page) -> str | None: for selector in ( ".ant-message-error", ".ant-form-item-explain-error", ".ant-alert-error", ): try: locator = page.locator(selector).first if locator.is_visible(timeout=300): text = locator.inner_text(timeout=1_000).strip() if text: return text except Exception: continue return None def _pathname_matches(pattern: str, pathname: str) -> bool: pattern = pattern.strip() if pattern in {"**/app/**", "**/app/*", "/app/**"}: return pathname.startswith("/app") if pattern.endswith("/**"): prefix = pattern[:-3].rstrip("/") if prefix.startswith("**/"): prefix = prefix[3:] if not prefix.startswith("/"): prefix = "/" + prefix return pathname.startswith(prefix) if "**" in pattern or "*" in pattern: regex = "^" + re.escape(pattern).replace(r"\*\*", ".*").replace(r"\*", "[^/]*") + "$" return re.search(regex, pathname) is not None return pathname == pattern or pathname.startswith(pattern) def _wait_for_url_pattern(page, pattern: str, timeout: int = 60_000) -> None: """SPA 路由用 pathname 轮询；glob 模式不依赖 navigation 事件。""" deadline = time.monotonic() + timeout / 1000 last_error: str | None = None while time.monotonic() < deadline: pathname = page.evaluate("() => window.location.pathname") if _pathname_matches(pattern, pathname): try: page.wait_for_load_state("networkidle", timeout=8_000) except Exception: page.wait_for_timeout(800) return err = _page_error_text(page) if err and err != last_error: last_error = err logger.warning("页面提示：%s", err) if "/login" in pathname: raise RuntimeError(f"登录失败：{err}") page.wait_for_timeout(400) err = _page_error_text(page) hint_parts = [f"当前 URL：`{page.url}`"] if err: hint_parts.append(f"页面错误：{err}") elif last_error: hint_parts.append(f"页面错误：{last_error}") hint_parts.append("请确认 PREVIEW_LOGIN_USER/PASSWORD 正确，且登录 API（内网网关）可达。") raise RuntimeError(f"等待 URL 匹配 `{pattern}` 超时（{timeout}ms）。{' '.join(hint_parts)}") def _click_target(page, target: str) -> None: target = interpolate(target.strip()) if target.lower() in {"登录", "login"}: for selector in ("button.login-submit", "button[type='submit']"): locator = page.locator(selector) if locator.count() > 0: locator.first.click(timeout=10_000) return candidates = [ page.get_by_role("menuitem", name=target, exact=True), page.get_by_role("button", name=target, exact=True), page.get_by_role("link", name=target, exact=True), page.get_by_text(target, exact=True), page.get_by_text(target, exact=False), ] for locator in candidates: if locator.count() > 0: locator.first.click(timeout=10_000) return raise RuntimeError(f"未找到可点击元素：{target}") def _execute_step(page, base_url: str, step: dict[str, Any]) -> None: action = str(step.get("action", "")).lower() if action == "goto": target = step.get("target") or step.get("url") or "/" url = resolve_url(base_url, str(target)) page.goto(url, wait_until="networkidle", timeout=60_000) return if action == "fill": field = str(step.get("field") or step.get("target") or "账号") _fill_field(page, field, step) return if action == "click": target = step.get("target") or step.get("text") if not target: raise RuntimeError("click 步骤缺少 target") _click_target(page, str(target)) page.wait_for_timeout(800) return if action == "wait": timeout = int(step.get("timeout") or 60_000) if step.get("url"): _wait_for_url_pattern(page, str(step["url"]), timeout=timeout) return if step.get("selector"): page.locator(interpolate(str(step["selector"]))).wait_for(timeout=30_000) return if step.get("text"): page.get_by_text(interpolate(str(step["text"])), exact=False).wait_for(timeout=30_000) return ms = int(step.get("ms") or 1500) page.wait_for_timeout(ms) return if action == "press": key = str(step.get("key") or step.get("target") or "Enter") page.keyboard.press(key) return if action == "screenshot": return raise RuntimeError(f"未知步骤 action={action}") def run_browser_scenario_sync(scenario: BrowserScenario) -> BrowserResult: from playwright.sync_api import sync_playwright stamp = datetime.now().strftime("%Y%m%d-%H%M%S") slug = (scenario.name or "browser").replace(" ", "-") output = SCREENSHOT_DIR / f"{slug}-{stamp}.png" output.parent.mkdir(parents=True, exist_ok=True) step_log: list[str] = [] final_url = scenario.base_url with sync_playwright() as playwright: browser = playwright.chromium.launch(headless=True) page = browser.new_page(viewport={"width": 1280, "height": 720}) steps = list(scenario.steps) if steps and steps[-1].get("action") != "screenshot" and not any( s.get("action") == "screenshot" for s in steps ): steps.append({"action": "screenshot"}) for index, step in enumerate(steps): label = _step_label(step, index) logger.info("执行步骤 %s", label) action = str(step.get("action", "")).lower() if action == "screenshot": page.wait_for_timeout(int(step.get("ms") or 1500)) page.screenshot(path=str(output), full_page=False, type="png") final_url = page.url step_log.append(label + " ✓") continue try: _execute_step(page, scenario.base_url, step) final_url = page.url step_log.append(label + " ✓") except Exception as exc: err = _page_error_text(page) detail = f"（{err}）" if err else "" raise RuntimeError(f"步骤失败：{label} @ {page.url}{detail}") from exc browser.close() return BrowserResult( scenario_name=scenario.name, base_url=scenario.base_url, final_url=final_url, screenshot_path=output, step_count=len(steps), step_log=step_log, )