270 lines
9.2 KiB
Python
270 lines
9.2 KiB
Python
"""通用 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,
|
||
)
|