提交 9a777cdc authored 作者: 陈泽健's avatar 陈泽健

feat(smart-locate): 智能定位升级为语义化用例生成器(阶段A-C完成并验证)

- 后端 smart_locate_service:语义解析优先路径 + 反向生成 semantic + 模式步骤锚点值作 text + 反向生成经执行期 resolve_step 校验(row_checkbox→row 降级)+ 动态值 CTX 化 + 全候选点击链
- routers/smart_locate:SmartLocateResult 增加 semantic/verify
- keyword_matcher:find_element_by_semantic 兜底 return None, []
- page_url_mapping:注册会议列表 scope
- 前端:batchLocate 回填 semantic/verify
- 单测 37 用例 + E2E 连跑两次 7/7 全绿 + PRD/执行计划文档
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 10ed34d7
......@@ -57,6 +57,19 @@
"fingerprint": [
{"selector": ".detail_top", "min_count": 1}
]
},
{
"id": "meeting_list",
"name": "会议列表",
"url": "https://192.168.5.44/#/meetingV3?meetingV3=%2FmeetingV3%2F%23%2FmeetingAll",
"url_patterns": [
"meetingV3.*meetingAll",
"meetingV3.*MeetingList",
"meeting/list"
],
"fingerprint": [
{"selector": ".meeting_list", "min_count": 1}
]
}
]
}
......@@ -58,6 +58,9 @@ class SmartLocateResult(BaseModel):
claude_enhanced: bool = Field(default=False, description="是否经过 Claude 语义增强")
claude_confidence: float = Field(default=0.0, description="Claude 置信度")
claude_reason: str = Field(default="", description="Claude 选择理由")
# ⚠️ 阶段6升级:语义化用例生成
semantic: Optional[Dict[str, Any]] = Field(None, description="生成的 semantic 语义目标")
verify: Optional[Dict[str, Any]] = Field(None, description="条件生成的 verify 状态验证")
class SmartLocateResponse(BaseModel):
......@@ -67,6 +70,8 @@ class SmartLocateResponse(BaseModel):
located_steps: int = Field(..., description="成功定位步骤数")
results: List[SmartLocateResult] = Field(default_factory=list, description="定位结果列表")
message: str = Field(default="", description="总体说明")
# ⚠️ 阶段6升级:动态数据 CTX 化收集的参数初值
parameters: Dict[str, str] = Field(default_factory=dict, description="动态数据参数初值(param{N}: 值)")
class VerifyStepRequest(BaseModel):
......@@ -142,16 +147,18 @@ async def smart_locate(
def _sync_locate():
"""同步执行智能定位(在线程池中运行)"""
service = SmartLocateService()
return service.locate_steps(
results = service.locate_steps(
steps=steps_data,
auto_login=request.auto_login,
navigate_menu=request.navigate_menu,
page_url=request.page_url,
use_claude=request.use_claude
)
# ⚠️ 阶段6升级:返回动态数据 CTX 化收集的参数初值
return results, dict(service.collected_parameters)
# 执行并等待结果
results = await loop.run_in_executor(None, _sync_locate)
results, parameters = await loop.run_in_executor(None, _sync_locate)
# 统计成功数量
located_count = sum(1 for r in results if r.get('success'))
......@@ -162,6 +169,7 @@ async def smart_locate(
total_steps=len(request.steps),
located_steps=located_count,
results=[SmartLocateResult(**r) for r in results],
parameters=parameters,
message=f"成功定位 {located_count}/{len(request.steps)} 个步骤"
)
......
......@@ -1177,6 +1177,9 @@ def find_element_by_semantic(
except Exception:
continue
# 兜底:未匹配到任何元素(此前返回 None 会导致调用方 tuple 解包崩溃)
return None, []
def resolve_selectors(
page,
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
部署智能定位功能到 192.168.5.60 服务器
"""
import paramiko
import os
import sys
# 服务器配置
SERVER = "192.168.5.60"
USER = "ubains"
PASSWORD = "Ubains@123"
DEPLOY_DIR = "/data/third_party/plat-auto-test"
# 需要上传的文件列表
FILES_TO_UPLOAD = [
# 新增文件
("backend/app/models/login_template.py", "backend/app/models/login_template.py"),
("backend/app/routers/login_template.py", "backend/app/routers/login_template.py"),
("backend/app/routers/smart_locate.py", "backend/app/routers/smart_locate.py"),
("backend/app/schemas/login_template.py", "backend/app/schemas/login_template.py"),
("backend/app/services/keyword_matcher.py", "backend/app/services/keyword_matcher.py"),
("backend/app/services/login_template_service.py", "backend/app/services/login_template_service.py"),
("backend/app/services/selector_extractor.py", "backend/app/services/selector_extractor.py"),
("backend/app/services/smart_locate_service.py", "backend/app/services/smart_locate_service.py"),
# 修改的文件
("backend/app/main.py", "backend/app/main.py"),
("backend/app/models/__init__.py", "backend/app/models/__init__.py"),
("frontend/src/api/elementLocate.ts", "frontend/src/api/elementLocate.ts"),
("frontend/src/views/Cases.vue", "frontend/src/views/Cases.vue"),
]
def deploy():
"""部署到服务器"""
print("=" * 60)
print("部署智能定位功能到服务器")
print("=" * 60)
print(f"服务器: {SERVER}")
print(f"用户: {USER}")
print(f"部署目录: {DEPLOY_DIR}")
print()
# 连接服务器
print("连接服务器...")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(SERVER, username=USER, password=PASSWORD)
print("[OK] SSH 连接成功")
except Exception as e:
print(f"[FAIL] SSH 连接失败: {e}")
return False
sftp = ssh.open_sftp()
# 上传文件
print(f"\n上传 {len(FILES_TO_UPLOAD)} 个文件...")
local_base = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
success_count = 0
for local_path, remote_path in FILES_TO_UPLOAD:
full_local = os.path.join(local_base, local_path)
full_remote = f"{DEPLOY_DIR}/{remote_path}"
try:
# 确保远程目录存在
remote_dir = os.path.dirname(full_remote)
try:
sftp.stat(remote_dir)
except FileNotFoundError:
ssh.exec_command(f"mkdir -p {remote_dir}")
# 上传文件
sftp.put(full_local, full_remote)
print(f" [OK] {remote_path}")
success_count += 1
except Exception as e:
print(f" [FAIL] {remote_path}: {e}")
sftp.close()
print(f"\n上传完成: {success_count}/{len(FILES_TO_UPLOAD)} 成功")
# 重启容器
print("\n重启 Docker 容器...")
stdin, stdout, stderr = ssh.exec_command(f"cd {DEPLOY_DIR}/deploy && docker compose restart app")
print(stdout.read().decode())
print(stderr.read().decode())
# 等待服务启动
print("等待服务启动...")
import time
time.sleep(5)
# 验证部署
print("\n验证部署...")
stdin, stdout, stderr = ssh.exec_command("curl -s http://localhost:8001/api/element/health")
result = stdout.read().decode()
if "smart-locate" in result or "healthy" in result:
print("[OK] 智能定位 API 已部署")
else:
print(f"[WARN] 验证结果: {result}")
stdin, stdout, stderr = ssh.exec_command("curl -s http://localhost:8001/api/login-templates")
result = stdout.read().decode()
if "template" in result:
print("[OK] 登录模板 API 已部署")
else:
print(f"[WARN] 验证结果: {result[:200]}")
ssh.close()
print("\n部署完成!")
return True
if __name__ == "__main__":
deploy()
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
测试脚本:验证智能定位 API 端到端功能
测试内容:
1. 健康检查
2. 智能定位 - 信息发布模块导航验证
"""
import requests
import json
import sys
BASE_URL = "http://localhost:8002"
def test_health():
"""测试健康检查"""
print("\n1. 健康检查...")
resp = requests.get(f"{BASE_URL}/api/element/health")
print(f" Status: {resp.status_code}")
if resp.status_code == 200:
data = resp.json()
print(f" Service: {data.get('service', 'N/A')}")
print(f" Status: {data.get('status', 'N/A')}")
else:
print(f" Error: {resp.text}")
def test_smart_locate():
"""测试智能定位"""
print("\n2. 智能定位 - 信息发布模块...")
# 构造请求:模拟用户录入的自然语言步骤
# 注意:auto_login=True 时,系统会自动执行登录模板
# 所以用户步骤只需要从登录后的操作开始
request_data = {
"steps": [
# 登录后的操作步骤(登录由 auto_login 自动完成)
{"order": 1, "name": "点击新增按钮", "action": "click", "params": {}},
{"order": 2, "name": "等待页面加载", "action": "wait", "params": {"timeout": 10000}},
],
"auto_login": True,
"navigate_menu": "信息发布",
"page_url": "https://192.168.5.44"
}
print(f" Request: {json.dumps(request_data, ensure_ascii=False, indent=2)}")
print(" Sending request (this may take 30-60 seconds)...")
try:
resp = requests.post(
f"{BASE_URL}/api/element/smart-locate",
json=request_data,
timeout=120 # 2 分钟超时
)
print(f" Status: {resp.status_code}")
if resp.status_code == 200:
data = resp.json()
print(f" Success: {data.get('success')}")
print(f" Total: {data.get('total_steps')}")
print(f" Located: {data.get('located_steps')}")
print(f" Message: {data.get('message')}")
for result in data.get('results', []):
status = "[OK]" if result.get('success') else "[FAIL]"
selector = result.get('params', {}).get('selector', 'N/A')
print(f" {status} Step {result.get('order')}: {result.get('name')} -> {selector}")
if result.get('message'):
print(f" Message: {result.get('message')}")
else:
print(f" Error: {resp.text}")
except requests.exceptions.Timeout:
print(" [TIMEOUT] Request timed out after 120 seconds")
except Exception as e:
print(f" [ERROR] {e}")
def test_login_templates():
"""测试登录模板 API"""
print("\n3. 登录模板 API...")
resp = requests.get(f"{BASE_URL}/api/login-templates")
print(f" Status: {resp.status_code}")
if resp.status_code == 200:
data = resp.json()
print(f" Total: {data.get('total')}")
for item in data.get('items', []):
print(f" - {item.get('name')} (type={item.get('template_type')}, steps={len(item.get('steps', []))})")
else:
print(f" Error: {resp.text}")
if __name__ == "__main__":
print("=" * 60)
print("智能定位 API 端到端测试")
print("=" * 60)
test_health()
test_login_templates()
test_smart_locate()
print("\n" + "=" * 60)
print("测试完成")
print("=" * 60)
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
测试智能定位 API 端到端流程
"""
import requests
import json
BASE_URL = "http://localhost:8001"
def test_smart_locate_api():
"""测试智能定位 API"""
print("\n" + "=" * 60)
print("测试智能定位 API")
print("=" * 60)
# 测试步骤:会议管理 - 查看列表
test_steps = [
{"order": 1, "name": "点击新增按钮", "action": "click", "params": {}},
{"order": 2, "name": "等待对话框加载", "action": "wait", "params": {"timeout": 5000}},
{"order": 3, "name": "输入会议主题", "action": "fill", "params": {"value": "测试会议"}},
{"order": 4, "name": "点击确定按钮", "action": "click", "params": {}},
]
payload = {
"steps": test_steps,
"auto_login": True,
"navigate_menu": "会议管理",
"page_url": "https://192.168.5.44"
}
print("\n请求参数:")
print(json.dumps(payload, indent=2, ensure_ascii=False))
print("\n发送请求...")
try:
response = requests.post(
f"{BASE_URL}/api/element/smart-locate",
json=payload,
timeout=180
)
print(f"响应状态码: {response.status_code}")
if response.status_code == 200:
result = response.json()
print("\n响应结果:")
print(json.dumps(result, indent=2, ensure_ascii=False))
print("\n定位汇总:")
print(f" 总步骤: {result.get('total_steps', 0)}")
print(f" 成功定位: {result.get('located_steps', 0)}")
for r in result.get("results", []):
status = "[OK]" if r.get("success") else "[FAIL]"
print(f" {status} 步骤 {r['order']}: {r['name']}")
if r.get("success"):
print(f" 主选择器: {r['selectors'].get('primary', 'N/A')}")
else:
print(f" 失败原因: {r.get('message', 'N/A')}")
return result.get("located_steps", 0) > 0
else:
print(f"请求失败: {response.text}")
return False
except Exception as e:
print(f"请求异常: {e}")
return False
if __name__ == "__main__":
test_smart_locate_api()
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_smart_locate_e2e.py
模块描述:智能定位端到端测试脚本
作者:czj
创建日期:2026-08-05
最后修改:2026-08-05
"""
import sys
import os
# 添加 backend 目录到 Python 路径
backend_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, backend_dir)
import asyncio
from app.services.smart_locate_service import SmartLocateService
from app.services.login_template_service import LoginTemplateService
from app.database import get_db, async_session_maker
async def test_login_templates():
"""测试登录模板服务"""
print("\n" + "=" * 60)
print("测试登录模板服务")
print("=" * 60)
async with async_session_maker() as db:
service = LoginTemplateService(db)
# 获取模板列表
templates, total = await service.get_list()
print(f"\n[OK] 模板列表: {total} 个模板")
for t in templates:
steps_count = len(t.steps) if t.steps else 0
print(f" - {t.name} ({t.template_type}): {steps_count} 步骤")
# 获取默认登录模板
login_template = await service.get_default_template("login")
if login_template:
print(f"\n[OK] 默认登录模板: {login_template.name}")
print(f" 步骤数: {len(login_template.steps)}")
for i, step in enumerate(login_template.steps, 1):
print(f" {i}. {step.get('name', 'N/A')} - {step.get('action', 'N/A')}")
# 获取默认导航模板
nav_template = await service.get_default_template("navigate")
if nav_template:
print(f"\n[OK] 默认导航模板: {nav_template.name}")
print(f" 步骤数: {len(nav_template.steps)}")
return True
async def test_smart_locate():
"""测试智能定位服务"""
print("\n" + "=" * 60)
print("测试智能定位服务")
print("=" * 60)
# 测试用例:会议管理 - 新增会议
test_steps = [
{"order": 1, "name": "点击新增按钮", "action": "click", "params": {}},
{"order": 2, "name": "等待对话框加载", "action": "wait", "params": {"timeout": 5000}},
{"order": 3, "name": "输入会议主题", "action": "fill", "params": {"value": "测试会议"}},
{"order": 4, "name": "点击确定按钮", "action": "click", "params": {}},
]
print("\n测试步骤:")
for step in test_steps:
print(f" {step['order']}. {step['name']} ({step['action']})")
service = SmartLocateService()
try:
# locate_steps 是同步方法,需要在线程池中运行
import asyncio
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None,
service.locate_steps,
test_steps,
True, # auto_login
"会议管理", # navigate_menu
"https://192.168.5.44" # page_url
)
print(f"\n定位结果:")
print(f" 总步骤: {len(result)}")
located_steps = sum(1 for r in result if r.get("success"))
print(f" 成功定位: {located_steps}")
for r in result:
status = "[OK]" if r.get("success") else "[FAIL]"
print(f"\n {status} 步骤 {r['order']}: {r['name']}")
if r.get("success"):
print(f" 主选择器: {r['selectors'].get('primary', 'N/A')}")
else:
print(f" 失败原因: {r.get('message', 'N/A')}")
return located_steps > 0
except Exception as e:
print(f"\n[ERROR] 测试失败: {str(e)}")
import traceback
traceback.print_exc()
return False
async def main():
"""主测试流程"""
print("\n" + "=" * 60)
print("智能定位端到端测试")
print("=" * 60)
print(f"被测系统: https://192.168.5.44")
print(f"测试菜单: 会议管理")
print(f"验证码: csba (固定)")
results = []
# 测试1: 登录模板
try:
r1 = await test_login_templates()
results.append(("登录模板服务", r1))
except Exception as e:
print(f"\n[ERROR] 登录模板测试失败: {e}")
results.append(("登录模板服务", False))
# 测试2: 智能定位
try:
r2 = await test_smart_locate()
results.append(("智能定位服务", r2))
except Exception as e:
print(f"\n[ERROR] 智能定位测试失败: {e}")
results.append(("智能定位服务", False))
# 汇总结果
print("\n" + "=" * 60)
print("测试汇总")
print("=" * 60)
for name, passed in results:
status = "[OK] 通过" if passed else "[FAIL] 失败"
print(f" {name}: {status}")
all_passed = all(r[1] for r in results)
print("\n" + ("全部测试通过!" if all_passed else "部分测试失败"))
return all_passed
if __name__ == "__main__":
asyncio.run(main())
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:verify_smart_locate_semantic.py
模块描述:智能定位语义化生成端到端验证脚本(阶段6语义化用例生成器)
在真实被测系统上运行 SmartLocateService.locate_steps,验证:
- 静态步骤走语义解析优先路径(产出 semantic)
- "xxx选择/切换为:值" 模式步骤产出 row_checkbox/tab semantic + contains CTX
- 动态 fill 步骤值 CTX 化 → parameters 初值
- 点击弹窗步骤条件生成 verify
- 全流程不产生业务数据副作用(在"确定创建"之前停止)
用法:
cd backend && python scripts/verify_smart_locate_semantic.py
作者:czj
创建日期:2026-08-18
"""
import io
import logging
import sys
from typing import Any, Dict, List
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(message)s")
logger = logging.getLogger("verify_semantic")
sys.path.insert(0, ".")
def main() -> None:
from app.services.smart_locate_service import SmartLocateService
steps: List[Dict[str, Any]] = [
# ===== 导航步骤(显式稳定选择器,聚焦语义生成验收而非导航智能) =====
# 名称避开 dialog 关键词(新建/创建/新增…),否则 _execute_step 会等待弹窗而
# 实际是页面跳转(CreateMeeting 为整页路由,无 .el-dialog/.el-drawer)→ 假失败
{"order": 1, "name": "点击【功能中心】展开", "action": "click",
"params": {"selector": '//*[@id="Home"]/div[1]/div[1]'}},
{"order": 2, "name": "进入会议模块", "action": "click",
"params": {"selector": '.el-drawer [data-key="reserve_list.create"]'}},
{"order": 3, "name": "等待会议表单加载", "action": "wait",
"params": {"selector": "input[placeholder*='会议名称']", "timeout": 10000}},
# ===== 表单步骤(无选择器 → 智能定位语义化生成验收核心) =====
# 4/7 动态 fill:keyword 定位 → 反向生成 input semantic + 值 CTX 化 → parameters 初值
# 5/6 "xxx:值" 模式 click:keyword 定位 → 反向生成 row_checkbox/tab semantic
# + contains CTX 化
{"order": 4, "name": "填写会议名称", "action": "fill",
"params": {"value": "智能定位验收自动会议X"}},
{"order": 5, "name": "会议室选择:北京展厅会议室", "action": "click", "params": {}},
{"order": 6, "name": "时间切换为:预定会议", "action": "click", "params": {}},
{"order": 7, "name": "输入会议议题", "action": "fill",
"params": {"value": "智能定位验收测试议题"}},
]
logger.info("=== 启动智能定位(真实系统 %s)===", "https://192.168.5.44")
service = SmartLocateService()
# navigate_menu 留空:由步骤自身完成"功能中心→会议模块"导航,
# 避免与 navigate_menu 的预导航冲突(后者已展开菜单,再点功能中心会失效)
results = service.locate_steps(
steps=steps,
auto_login=True,
navigate_menu="",
page_url="https://192.168.5.44",
use_claude=False, # 规则优先,验证语义生成而非 Claude
)
out: List[str] = []
out.append("")
out.append("=== 定位结果 ===")
for r in results:
ok = r.get("success")
sem = r.get("semantic")
ver = r.get("verify")
out.append(
f"#{r.get('order')} {r.get('name')}"
f" | action={r.get('action')} | {'OK' if ok else 'FAIL'}"
)
if ok and r.get('selectors', {}).get('primary'):
out.append(f" selector: {r['selectors']['primary']}")
if sem:
out.append(f" semantic: {sem}")
if ver:
out.append(f" verify: {ver}")
if ok and r.get('params', {}).get('value'):
out.append(f" params.value: {r['params']['value']}")
out.append("")
out.append("=== 动态数据参数初值 (collected_parameters) ===")
for k, v in service.collected_parameters.items():
out.append(f" {k}: {v}")
# ===== 断言汇总 =====
ok_count = sum(1 for r in results if r.get('success'))
sem_count = sum(1 for r in results if r.get('semantic'))
out.append("")
out.append(f"=== 统计: 成功 {ok_count}/{len(results)}, 含 semantic {sem_count} ===")
io.open("_verify_semantic_out.txt", "w", encoding="utf-8").write("\n".join(out))
logger.info("输出已写入 _verify_semantic_out.txt")
for line in out:
print(line)
if __name__ == "__main__":
main()
此差异已折叠。
......@@ -180,6 +180,10 @@ export interface SmartLocateResult {
screenshot?: string
/** 说明信息 */
message: string
/** 生成的 semantic 语义目标(阶段6语义化用例生成器) */
semantic?: Record<string, any> | null
/** 条件生成的 verify 状态验证 */
verify?: Record<string, any> | null
}
/**
......@@ -196,6 +200,8 @@ export interface SmartLocateResponse {
results: SmartLocateResult[]
/** 总体说明 */
message: string
/** 动态数据参数初值(param{N}: 值,阶段6 CTX 化收集) */
parameters?: Record<string, string>
}
/**
......
......@@ -1123,19 +1123,30 @@ const batchLocate = async () => {
if (locateResult.selectors?.primary) {
updatedParams.selector = locateResult.selectors.primary
}
// ⚠️ 阶段6升级:合并 semantic / verify(新生成的优先,保留已有字段)
return {
...s,
params: updatedParams,
locator_type: 'css',
locator_value: locateResult.selectors?.primary || '',
semantic: locateResult.semantic ?? s.semantic,
verify: locateResult.verify ?? s.verify,
}
}
return s
})
// ⚠️ 阶段6升级:动态数据 CTX 化收集的参数初值合并进用例参数
// (param{N} 键与步骤内 {__CTX:param{N}__} 占位对齐,执行期才能解析)
const mergedParameters = {
...(caseDetail.parameters || {}),
...(result.parameters || {}),
}
// 5. 更新用例到数据库
await caseApi.update(locateConfigCaseId.value, {
steps: updatedSteps,
parameters: mergedParameters,
})
let msg = `智能定位完成:成功 ${result.located_steps}/${result.total_steps} 个步骤`
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论