308 lines
10 KiB
Python
308 lines
10 KiB
Python
"""解析自然语言 / YAML / 场景名 → 浏览器步骤。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import yaml
|
||
|
||
import env_config
|
||
from browser_env import default_base_url, interpolate
|
||
from browser_models import BrowserScenario
|
||
|
||
SCENARIO_DIRS = [
|
||
Path(__file__).resolve().parent / "scenarios",
|
||
Path(__file__).resolve().parent.parent / "scenarios",
|
||
]
|
||
|
||
|
||
def _project_cwd() -> Path:
|
||
raw = env_config.env("CURSOR_CWD", r"d:\LY\test\tech") or r"d:\LY\test\tech"
|
||
return Path(raw).resolve()
|
||
|
||
|
||
def _strip_mention(text: str) -> str:
|
||
return re.sub(r"@\S+\s*", "", text).strip()
|
||
|
||
|
||
def _scenario_search_dirs() -> list[Path]:
|
||
dirs = list(SCENARIO_DIRS)
|
||
dirs.append(_project_cwd() / ".browser-scenarios")
|
||
custom = (env_config.env("BROWSER_SCENARIOS_DIR") or "").strip()
|
||
if custom:
|
||
dirs.append(Path(custom).resolve())
|
||
return dirs
|
||
|
||
|
||
def is_browser_intent(text: str) -> bool:
|
||
raw = _strip_mention(text)
|
||
if not raw:
|
||
return False
|
||
if re.match(r"^(browser|网页|网页操作|操作)\b", raw, re.IGNORECASE):
|
||
return True
|
||
if re.search(r"```(?:yaml|yml)", raw, re.IGNORECASE):
|
||
return True
|
||
if re.search(r"(?m)^browser\s*:", raw, re.IGNORECASE):
|
||
return True
|
||
|
||
if re.match(r"^(preview|截图|预览|截屏)\s", raw, re.IGNORECASE):
|
||
if not re.search(r"[，,。；;]|然后|输入|点击|填写|访问|打开|登录", raw):
|
||
return False
|
||
|
||
if len(_split_segments(text)) >= 2:
|
||
return True
|
||
|
||
verbs = 0
|
||
for pattern in (r"访问", r"打开", r"输入", r"填写", r"点击", r"点选", r"选择", r"登录"):
|
||
if re.search(pattern, raw):
|
||
verbs += 1
|
||
return verbs >= 2
|
||
|
||
|
||
def _load_yaml_scenario(path: Path) -> BrowserScenario:
|
||
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
||
if not isinstance(data, dict):
|
||
raise RuntimeError(f"场景文件格式错误：{path}")
|
||
base_url = interpolate(str(data.get("base_url") or default_base_url()))
|
||
steps = data.get("steps")
|
||
if not isinstance(steps, list) or not steps:
|
||
raise RuntimeError(f"场景缺少 steps：{path}")
|
||
return BrowserScenario(
|
||
name=data.get("name") or path.stem,
|
||
base_url=base_url,
|
||
steps=_normalize_steps(steps),
|
||
source=f"file:{path.name}",
|
||
)
|
||
|
||
|
||
def _find_scenario_file(name: str) -> Path | None:
|
||
slug = name.strip().replace(" ", "-")
|
||
for directory in _scenario_search_dirs():
|
||
for candidate in (directory / f"{slug}.yaml", directory / f"{slug}.yml"):
|
||
if candidate.exists():
|
||
return candidate
|
||
return None
|
||
|
||
|
||
def _normalize_steps(raw_steps: list[Any]) -> list[dict[str, Any]]:
|
||
normalized: list[dict[str, Any]] = []
|
||
for item in raw_steps:
|
||
if isinstance(item, str):
|
||
normalized.append({"action": item})
|
||
continue
|
||
if not isinstance(item, dict) or not item:
|
||
raise RuntimeError(f"无效步骤：{item!r}")
|
||
if "action" in item:
|
||
normalized.append(dict(item))
|
||
continue
|
||
if len(item) == 1:
|
||
action, payload = next(iter(item.items()))
|
||
step = {"action": action}
|
||
if payload is not None:
|
||
if isinstance(payload, dict):
|
||
step.update(payload)
|
||
elif action == "wait" and isinstance(payload, int):
|
||
step["ms"] = payload
|
||
elif action == "wait" and isinstance(payload, str) and payload.isdigit():
|
||
step["ms"] = int(payload)
|
||
else:
|
||
step["target"] = payload
|
||
normalized.append(step)
|
||
continue
|
||
raise RuntimeError(f"无效步骤：{item!r}")
|
||
return normalized
|
||
|
||
|
||
def _parse_inline_dsl(text: str) -> BrowserScenario | None:
|
||
raw = _strip_mention(text)
|
||
match = re.search(r"(?ms)^browser\s*:\s*\n(.+)$", raw, re.IGNORECASE)
|
||
if not match:
|
||
return None
|
||
|
||
steps: list[dict[str, Any]] = []
|
||
for line in match.group(1).splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
line = re.sub(r"^[-*]\s*", "", line)
|
||
if not line:
|
||
continue
|
||
steps.append(_parse_dsl_line(line))
|
||
|
||
if not steps:
|
||
return None
|
||
return BrowserScenario(
|
||
name="inline",
|
||
base_url=default_base_url(),
|
||
steps=steps,
|
||
source="inline-dsl",
|
||
)
|
||
|
||
|
||
def _parse_dsl_line(line: str) -> dict[str, Any]:
|
||
parts = line.split(None, 2)
|
||
action = parts[0].lower()
|
||
if action == "goto":
|
||
return {"action": "goto", "target": parts[1] if len(parts) > 1 else "/"}
|
||
if action == "click":
|
||
return {"action": "click", "target": " ".join(parts[1:])}
|
||
if action == "fill":
|
||
if len(parts) < 3:
|
||
raise RuntimeError(f"fill 语法：fill 字段 值（{line}）")
|
||
return {"action": "fill", "field": parts[1], "value": parts[2]}
|
||
if action == "wait":
|
||
payload = parts[1] if len(parts) > 1 else "1500"
|
||
if payload.isdigit():
|
||
return {"action": "wait", "ms": int(payload)}
|
||
return {"action": "wait", "url": payload}
|
||
if action in {"screenshot", "shot"}:
|
||
return {"action": "screenshot"}
|
||
raise RuntimeError(f"未知 DSL 步骤：{line}")
|
||
|
||
|
||
def _split_segments(text: str) -> list[str]:
|
||
raw = _strip_mention(text)
|
||
raw = re.sub(r"^(browser|网页|网页操作|操作)\s*[:：]?\s*", "", raw, flags=re.IGNORECASE)
|
||
raw = re.sub(r"然后截图|再截图|最后截图", "截图", raw)
|
||
chunks = re.split(r"[，,。；;]\s*|\s+然后\s+|\s+接着\s+|\s+并\s*", raw)
|
||
expanded: list[str] = []
|
||
for chunk in chunks:
|
||
chunk = chunk.strip()
|
||
if not chunk:
|
||
continue
|
||
subchunks = re.split(r"\s+然后\s+", chunk)
|
||
if "后" in chunk and len(subchunks) == 1:
|
||
subchunks = re.split(r"(?<=[登录页表单])后(?=[进入打开等待点击访问])", chunk)
|
||
for part in subchunks:
|
||
part = part.strip()
|
||
if part:
|
||
expanded.append(part)
|
||
return expanded
|
||
|
||
|
||
def _parse_segment(segment: str) -> list[dict[str, Any]]:
|
||
seg = segment.strip()
|
||
if not seg or seg.lower() in {"browser", "网页操作"}:
|
||
return []
|
||
|
||
if re.fullmatch(r"截图|截屏", seg, re.IGNORECASE):
|
||
return [{"action": "screenshot"}]
|
||
|
||
match = re.search(r"访问登录页|打开登录页|进入登录页|运行登录页", seg, re.IGNORECASE)
|
||
if match:
|
||
return [{"action": "goto", "target": "/login"}]
|
||
|
||
match = re.search(r"输入账号密码|填写账号密码|输入账号和密码", seg, re.IGNORECASE)
|
||
if match:
|
||
return [
|
||
{"action": "fill", "field": "账号", "value": "{{PREVIEW_LOGIN_USER}}"},
|
||
{"action": "fill", "field": "密码", "value": "{{PREVIEW_LOGIN_PASSWORD}}"},
|
||
]
|
||
|
||
match = re.search(r"输入账号|填写账号|输入用户名|填写用户名", seg, re.IGNORECASE)
|
||
if match:
|
||
return [{"action": "fill", "field": "账号", "value": "{{PREVIEW_LOGIN_USER}}"}]
|
||
|
||
match = re.search(r"输入密码|填写密码", seg, re.IGNORECASE)
|
||
if match:
|
||
return [{"action": "fill", "field": "密码", "value": "{{PREVIEW_LOGIN_PASSWORD}}"}]
|
||
|
||
match = re.search(r"进入主页|进入首页|打开主页|打开首页|等待主页", seg, re.IGNORECASE)
|
||
if match:
|
||
return [{"action": "wait", "url": "**/app/**"}]
|
||
|
||
match = re.search(r"等待\s*(\d+)\s*秒", seg, re.IGNORECASE)
|
||
if match:
|
||
return [{"action": "wait", "ms": int(match.group(1)) * 1000}]
|
||
|
||
match = re.search(
|
||
r"(?:点击|点选|选择)\s*(.+?)(?:菜单|按钮|链接)?$",
|
||
seg,
|
||
re.IGNORECASE,
|
||
)
|
||
if match:
|
||
target = match.group(1).strip()
|
||
target = re.sub(r"(然后|再|并)?\s*(截图|截屏).*$", "", target, flags=re.IGNORECASE).strip()
|
||
target = re.sub(r"(然后|再|之后)$", "", target).strip()
|
||
target = re.sub(r"(菜单|按钮|链接)$", "", target).strip()
|
||
if target:
|
||
return [{"action": "click", "target": target}]
|
||
|
||
match = re.search(
|
||
r"(?:访问|打开|进入)\s*(https?://\S+|/\S+|登录页|主页|首页)",
|
||
seg,
|
||
re.IGNORECASE,
|
||
)
|
||
if match:
|
||
target = match.group(1)
|
||
mapping = {"登录页": "/login", "主页": "/app/dashboard", "首页": "/app/dashboard"}
|
||
return [{"action": "goto", "target": mapping.get(target, target)}]
|
||
|
||
return []
|
||
|
||
|
||
def parse_natural_language(text: str) -> BrowserScenario | None:
|
||
segments = _split_segments(text)
|
||
steps: list[dict[str, Any]] = []
|
||
for segment in segments:
|
||
steps.extend(_parse_segment(segment))
|
||
|
||
if not steps:
|
||
return None
|
||
if not any(step.get("action") == "screenshot" for step in steps):
|
||
if re.search(r"截图|截屏", text, re.IGNORECASE):
|
||
steps.append({"action": "screenshot"})
|
||
if not steps:
|
||
return None
|
||
|
||
return BrowserScenario(
|
||
name="natural",
|
||
base_url=default_base_url(),
|
||
steps=steps,
|
||
source="natural-language",
|
||
)
|
||
|
||
|
||
def parse_browser_request(text: str) -> BrowserScenario | None:
|
||
if not is_browser_intent(text):
|
||
return None
|
||
|
||
raw = _strip_mention(text)
|
||
yaml_block = re.search(r"```(?:yaml|yml)\s*\n(.+?)```", raw, re.IGNORECASE | re.DOTALL)
|
||
if yaml_block:
|
||
data = yaml.safe_load(yaml_block.group(1))
|
||
if isinstance(data, dict):
|
||
base_url = interpolate(str(data.get("base_url") or default_base_url()))
|
||
steps = data.get("steps") or []
|
||
return BrowserScenario(
|
||
name=data.get("name") or "yaml-inline",
|
||
base_url=base_url,
|
||
steps=_normalize_steps(steps),
|
||
source="yaml-inline",
|
||
)
|
||
|
||
inline = _parse_inline_dsl(text)
|
||
if inline:
|
||
return inline
|
||
|
||
match = re.match(r"^(browser|网页|网页操作|操作)\s+([\w\-./]+)\s*$", raw, re.IGNORECASE)
|
||
if match:
|
||
path = _find_scenario_file(match.group(2))
|
||
if not path:
|
||
raise RuntimeError(f"未找到场景文件：{match.group(2)}.yaml")
|
||
return _load_yaml_scenario(path)
|
||
|
||
scenario = parse_natural_language(text)
|
||
if scenario:
|
||
return scenario
|
||
|
||
default_name = (env_config.env("BROWSER_DEFAULT_SCENARIO") or "").strip()
|
||
if default_name:
|
||
path = _find_scenario_file(default_name)
|
||
if path:
|
||
return _load_yaml_scenario(path)
|
||
|
||
return None
|