项目初始化
This commit is contained in:
307
bot/browser_parser.py
Normal file
307
bot/browser_parser.py
Normal file
@@ -0,0 +1,307 @@
|
||||
"""解析自然语言 / 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
|
||||
Reference in New Issue
Block a user