Skip to content

U
ubains-module-test
提交 b2854e30 authored 作者: 陈泽健's avatar 陈泽健
浏览文件
操作

fix(ui-cases): 修复步骤表格不显示定位信息问题 

- Cases.vue 新增"定位类型"和"定位值"两列
- 兼容多种命名格式（locatorType/locator_type/params.selector）
- formatParams 排除 selector 字段避免重复显示
- element_locator.py 批量定位同时写入驼峰和蛇形命名
- 新增批量获取定位按钮（绿色），一键获取用例所有步骤定位
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 60ae5431
隐藏空白字符变更
内嵌 并排
正在显示 包含 482 行增加 和 279 行删除
backend/app/routers/element_locator.py
浏览文件 @ b2854e30
......@@ -3,17 +3,22 @@
提供本地元素定位功能，通过 Playwright 访问被测页面，
提取可交互元素，匹配步骤描述，返回候选定位器列表。
支持单个步骤定位和批量定位（一键获取用例所有步骤的定位）。
@author czj
@date 2026-08-03
"""
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
import logging
import asyncio
import threading
import subprocess
import json
import os
import sys
from app.executors.playwright_executor import PlaywrightExecutor
from app.config import settings
......@@ -26,18 +31,18 @@ router = APIRouter(tags=["元素定位"])
class LocateRequest(BaseModel):
"""元素定位请求"""
step_description: str # 步骤描述，如"输入用户名"
page_url: str # 目标页面 URL
expected: str = "" # 预期结果（可选）
auto_login: bool = True # 是否自动登录（默认 True）
step_description: str
page_url: str
expected: str = ""
auto_login: bool = True
class LocateCandidate(BaseModel):
"""定位候选"""
locator_type: str # css/xpath/text/id/name
locator_value: str # 定位值
confidence: float # 置信度 0-1
element_info: Dict[str, Any] # 元素信息
locator_type: str
locator_value: str
confidence: float
element_info: Dict[str, Any]
class LocateResponse(BaseModel):
......@@ -45,102 +50,83 @@ class LocateResponse(BaseModel):
success: bool
candidates: List[LocateCandidate]
message: str
screenshot: Optional[str] = None # 页面截图（base64）
screenshot: Optional[str] = None
# ==================== 元素提取函数 ====================
class BatchLocateRequest(BaseModel):
"""批量元素定位请求"""
case_id: str
page_url: str = "https://192.168.5.44"
auto_login: bool = True
def extract_interactive_elements(page) -> list:
"""
提取页面中所有可交互元素的信息
支持 micro-app 微前端容器
Args:
page: Playwright Page 对象
class StepLocateResult(BaseModel):
"""单个步骤的定位结果"""
order: int
description: str
success: bool
locator_type: str = ""
locator_value: str = ""
confidence: float = 0.0
message: str = ""
Returns:
list: 元素信息列表
"""
class BatchLocateResponse(BaseModel):
"""批量元素定位响应"""
success: bool
case_id: str
total_steps: int
located_steps: int
results: List[StepLocateResult]
message: str
# ==================== 元素提取函数 ====================
def extract_interactive_elements(page) -> list:
"""提取页面中所有可交互元素的信息，支持 micro-app 微前端容器"""
return page.evaluate('''() => {
const elements = [];
// 1. 主文档元素
document.querySelectorAll('input,button,a,select,textarea,label,[role="button"]').forEach(el => {
elements.push({
tag: el.tagName,
type: el.type || '',
placeholder: el.placeholder || '',
text: (el.innerText || '').substring(0, 50),
id: el.id || '',
name: el.name || '',
className: el.className || '',
ariaLabel: el.getAttribute('aria-label') || '',
dataTestid: el.getAttribute('data-testid') || '',
href: el.href || '',
value: el.value || '',
tag: el.tagName, type: el.type || '', placeholder: el.placeholder || '',
text: (el.innerText || '').substring(0, 50), id: el.id || '', name: el.name || '',
className: el.className || '', ariaLabel: el.getAttribute('aria-label') || '',
dataTestid: el.getAttribute('data-testid') || '', href: el.href || '', value: el.value || '',
});
});
// 2. micro-app 微前端容器
document.querySelectorAll('micro-app').forEach(microApp => {
const shadowBody = microApp.querySelector('micro-app-body');
if (shadowBody && shadowBody.shadowRoot) {
shadowBody.shadowRoot.querySelectorAll('input,button,a,select,textarea').forEach(el => {
elements.push({
tag: el.tagName,
type: el.type || '',
placeholder: el.placeholder || '',
text: (el.innerText || '').substring(0, 50),
id: el.id || '',
name: el.name || '',
tag: el.tagName, type: el.type || '', placeholder: el.placeholder || '',
text: (el.innerText || '').substring(0, 50), id: el.id || '', name: el.name || '',
inMicroApp: true,
});
});
}
});
// 3. 普通 iframe（可选）
document.querySelectorAll('iframe').forEach(iframe => {
try {
const iframeDoc = iframe.contentDocument;
if (iframeDoc) {
iframeDoc.querySelectorAll('input,button,a,select,textarea').forEach(el => {
elements.push({
tag: el.tagName,
type: el.type || '',
placeholder: el.placeholder || '',
text: (el.innerText || '').substring(0, 50),
id: el.id || '',
name: el.name || '',
tag: el.tagName, type: el.type || '', placeholder: el.placeholder || '',
text: (el.innerText || '').substring(0, 50), id: el.id || '', name: el.name || '',
inIframe: true,
});
});
}
} catch (e) {
// 跨域 iframe 无法访问，忽略
}
} catch (e) {}
});
return elements;
}''')
def extract_keywords(text: str) -> list:
"""
从步骤描述中提取关键词
策略：
1. 先去除动作词（输入/点击/选择等），保留目标词
2. 将剩余文本按语义拆分为短词
3. 同时返回原始词和拆分后的子词
Args:
text: 步骤描述文本
Returns:
list: 关键词列表
"""
# 去除动作词
"""从步骤描述中提取关键词，支持中文语义拆分"""
action_words = ['输入', '填写', '填入', '键入', '写入', '点击', '按下', '选择',
'单击', '双击', '打开', '导航', '访问', '进入', '跳转', '前往',
'等待', '延时', '断言', '验证', '检查', '确认', '勾选', '取消',
......@@ -149,63 +135,30 @@ def extract_keywords(text: str) -> list:
cleaned = text
for action in sorted(action_words, key=len, reverse=True):
cleaned = cleaned.replace(action, '')
cleaned = cleaned.strip()
keywords = []
# 原始文本也加入（用于精确匹配）
keywords.append(text)
# 清理后的目标词
if cleaned and len(cleaned) > 0:
keywords = [text]
if cleaned:
keywords.append(cleaned)
# 将长词拆分为 2 字子词
if len(cleaned) >= 2:
for i in range(len(cleaned) - 1):
sub = cleaned[i:i+2]
if len(sub) == 2 and sub not in action_words:
keywords.append(sub)
# 单字也加入（用于更宽松的匹配）
for ch in cleaned:
if ch.strip() and ch not in ['的', '在', '是', '和', '有', '等', '中', '为', '了', '与', '或']:
keywords.append(ch)
# 去重
seen = set()
unique_keywords = []
for kw in keywords:
if kw not in seen:
seen.add(kw)
unique_keywords.append(kw)
return [kw for kw in keywords if kw not in seen and not seen.add(kw)]
return unique_keywords
def build_locator_candidates(step_description: str, elements: list, page_title: str = "") -> list:
"""
关键词算法匹配元素
返回候选定位器列表（按置信度降序）
Args:
step_description: 步骤描述
elements: 元素信息列表
page_title: 页面标题（用于上下文）
Returns:
list: 候选定位器列表
"""
def build_locator_candidates(step_description: str, elements: list, page_title: str = "", page_url: str = "") -> list:
"""关键词算法匹配元素，返回候选定位器列表（按置信度降序）"""
candidates = []
desc_lower = step_description.lower()
# 语义推断
is_input_action = any(kw in desc_lower for kw in ['输入', '填写', '键入', '写入', '填入'])
is_click_action = any(kw in desc_lower for kw in ['点击', '按下', '选择', '单击', '双击'])
is_navigate_action = any(kw in desc_lower for kw in ['打开', '导航', '访问', '进入', '跳转'])
# 提取关键词
keywords = extract_keywords(step_description)
logger.info(f"步骤描述: {step_description}, 关键词: {keywords}")
......@@ -214,229 +167,377 @@ def build_locator_candidates(step_description: str, elements: list, page_title:
locator_type = 'css'
locator_value = ''
# 1. ID 精确匹配（最高优先级）
if el.get('id'):
if any(kw in el['id'] for kw in keywords):
score += 0.4
locator_value = f"#{el['id']}"
# 2. placeholder 匹配
if el.get('placeholder'):
if any(kw in el['placeholder'] for kw in keywords):
score += 0.3
locator_value = f"[placeholder*='{el['placeholder']}']"
# 3. aria-label 匹配
if el.get('ariaLabel'):
if any(kw in el['ariaLabel'] for kw in keywords):
score += 0.3
locator_value = f"[aria-label*='{el['ariaLabel']}']"
# 4. 文本内容匹配（按钮/链接）
if el.get('text') and el['tag'] in ['BUTTON', 'A']:
if any(kw in el['text'] for kw in keywords):
score += 0.25
locator_value = f"{el['tag'].lower()}:has-text('{el['text']}')"
# 5. name 属性匹配
if el.get('name'):
if any(kw in el['name'] for kw in keywords):
score += 0.2
locator_value = f"[name='{el['name']}']"
# 6. data-testid 匹配
if el.get('dataTestid'):
if any(kw in el['dataTestid'] for kw in keywords):
score += 0.3
locator_value = f"[data-testid='{el['dataTestid']}']"
# 7. 语义推断加分
if el.get('id') and any(kw in el['id'] for kw in keywords):
score += 0.4; locator_value = f"#{el['id']}"
if el.get('placeholder') and any(kw in el['placeholder'] for kw in keywords):
score += 0.3; locator_value = f"[placeholder*='{el['placeholder']}']"
if el.get('ariaLabel') and any(kw in el['ariaLabel'] for kw in keywords):
score += 0.3; locator_value = f"[aria-label*='{el['ariaLabel']}']"
if el.get('text') and el['tag'] in ['BUTTON', 'A'] and any(kw in el['text'] for kw in keywords):
score += 0.25; locator_value = f"{el['tag'].lower()}:has-text('{el['text']}')"
if el.get('name') and any(kw in el['name'] for kw in keywords):
score += 0.2; locator_value = f"[name='{el['name']}']"
if el.get('dataTestid') and any(kw in el['dataTestid'] for kw in keywords):
score += 0.3; locator_value = f"[data-testid='{el['dataTestid']}']"
if is_input_action and el['tag'] in ['INPUT', 'TEXTAREA']:
score += 0.15
if is_click_action and el['tag'] in ['BUTTON', 'A']:
score += 0.15
if el.get('type') and el['tag'] == 'INPUT' and el['type'] in ['text', 'password', 'email', 'tel', 'number']:
if is_input_action:
score += 0.1
# 8. type 属性匹配（输入框）
if el.get('type') and el['tag'] == 'INPUT':
if el['type'] in ['text', 'password', 'email', 'tel', 'number']:
if is_input_action:
score += 0.1
# 只保留有 locator_value 且置信度 > 0.15 的候选
if locator_value and score > 0.15:
candidates.append({
'locator_type': locator_type,
'locator_value': locator_value,
'locator_type': locator_type, 'locator_value': locator_value,
'confidence': min(score, 1.0),
'element_info': {
'tag': el.get('tag', ''),
'type': el.get('type', ''),
'placeholder': el.get('placeholder', ''),
'text': el.get('text', ''),
'id': el.get('id', ''),
}
'element_info': {'tag': el.get('tag', ''), 'type': el.get('type', ''),
'placeholder': el.get('placeholder', ''), 'text': el.get('text', ''), 'id': el.get('id', '')}
})
# 按置信度降序排序
candidates.sort(key=lambda x: x['confidence'], reverse=True)
return candidates
# 如果最高置信度不足 0.6，通过 SSH 调用宿主机上的 ai_locate_worker.py
# 让宿主机直接用 Playwright 访问页面获取真实元素 + Claude CLI 分析
needs_claude = not candidates or candidates[0]['confidence'] < 0.6
if needs_claude:
logger.info("关键词算法置信度不足，调用宿主机 ai_locate_worker.py...")
claude_result = call_claude_code_via_ssh(step_description, page_url)
if claude_result and claude_result.get('locator_type') and claude_result.get('locator_value'):
confidence = claude_result.get('confidence', 0.8)
if confidence > 0.5:
is_duplicate = any(c['locator_value'] == claude_result['locator_value'] for c in candidates)
if not is_duplicate:
candidates.insert(0, {
'locator_type': claude_result['locator_type'],
'locator_value': claude_result['locator_value'],
'confidence': confidence,
'element_info': {'tag': '', 'type': '', 'placeholder': '', 'text': '', 'id': ''},
'_claude_reason': claude_result.get('reason', '')
})
logger.info(f"Claude Code 匹配: {claude_result['locator_value']} (置信度: {confidence})")
else:
logger.info("Claude Code 结果与已有候选重复，跳过")
else:
logger.info(f"Claude Code 置信度不足: {confidence}")
else:
logger.info("Claude Code 未返回有效结果")
return candidates[:5]
def call_claude_code_via_ssh(step_description: str, page_url: str, expected: str = "") -> dict:
"""
通过 SSH 调用宿主机上的 ai_locate_worker.py
宿主机直接使用 Playwright 访问页面获取真实元素 + Claude CLI 分析
利用已有的 ai_locate_worker.py 脚本（宿主机上已有），
它已经在宿主机上安装了 Playwright + Claude Code CLI。
"""
HOST = "192.168.5.60"
HOST_USER = "ubains"
SSH_PASSWORD = "Ubains@123"
DEPLOY_DIR = "/data/third_party/plat-auto-test"
# 在宿主机上执行 ai_locate_worker.py 脚本
# 这个脚本会：
# 1. Playwright 打开目标页面，提取真实 DOM 元素
# 2. 关键词算法 + Claude Code 精确匹配
# 3. 返回定位结果
cmd = (
f'cd {DEPLOY_DIR}/backend && '
f'python3 scripts/ai_locate_worker.py '
f'--step-desc "{step_description}" '
f'--page-url "{page_url}" '
f'--expected "{expected}"'
)
try:
logger.info(f"通过 SSH 调用宿主机 {HOST} 上的 ai_locate_worker.py...")
import paramiko
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(HOST, username=HOST_USER, password=SSH_PASSWORD, timeout=10)
stdin, stdout, stderr = ssh.exec_command(cmd, timeout=90)
output = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
ssh.close()
if output:
# 脚本输出中包含 JSON 结果
json_start = output.find('{')
json_end = output.rfind('}') + 1
if json_start >= 0 and json_end > json_start:
parsed = json.loads(output[json_start:json_end])
logger.info(f"ai_locate_worker.py 返回结果: success={parsed.get('success')}, candidates={len(parsed.get('candidates', []))}")
# 从结果中提取第一个候选（最优匹配）
candidates = parsed.get('candidates', [])
if candidates:
best = candidates[0]
return {
'locator_type': best.get('locator_type', ''),
'locator_value': best.get('locator_value', ''),
'confidence': best.get('confidence', 0.8),
'reason': best.get('reasons', ['Claude Code 分析匹配'])[0] if best.get('reasons') else 'Claude Code 分析匹配'
}
else:
logger.warning(f"脚本输出中未找到 JSON: {output[:300]}")
else:
logger.warning(f"SSH 命令无输出, stderr: {err[:200]}")
return {"locator_type": "", "locator_value": "", "confidence": 0, "reason": "Claude调用失败"}
except Exception as e:
logger.warning(f"Claude Code 调用异常: {e}")
return {"locator_type": "", "locator_value": "", "confidence": 0, "reason": f"Claude调用失败: {str(e)}"}
# ==================== API 端点 ====================
def _do_locate_with_executor(executor, page_url: str, auto_login: bool) -> dict:
"""使用已有的 PlaywrightExecutor 执行定位准备工作
返回 { success, elements, page_title, current_url, message }
"""
# 自动登录
if auto_login and not executor._is_logged_in:
logger.info("执行自动登录...")
login_success = executor.do_login()
if not login_success:
return {'success': False, 'elements': [], 'page_title': '', 'current_url': '', 'message': '自动登录失败'}
# 访问目标页面
logger.info(f"访问页面: {page_url}")
current_url = executor._page.url
is_login_page = current_url.rstrip('/') == 'https://192.168.5.44'
is_target_login = page_url.rstrip('/') == 'https://192.168.5.44' or 'login' in page_url.lower()
if executor._is_logged_in and is_target_login and not is_login_page:
logger.info(f"已在登录后页面，跳过导航。当前 URL: {current_url}")
else:
try:
executor._page.goto(page_url, wait_until="domcontentloaded", timeout=30000)
except Exception as e:
logger.warning(f"页面加载超时(networkidle): {e}")
try:
executor._page.goto(page_url, wait_until="domcontentloaded", timeout=15000)
except Exception as e2:
logger.warning(f"页面加载超时(domcontentloaded)，继续: {e2}")
executor._page.wait_for_timeout(3000)
# 提取元素
elements = extract_interactive_elements(executor._page)
page_title = executor._page.title()
current_url = executor._page.url
logger.info(f"提取到 {len(elements)} 个元素，页面标题: {page_title}")
return {'success': True, 'elements': elements, 'page_title': page_title, 'current_url': current_url, 'message': 'ok'}
# ==================== 单个步骤定位 API ====================
@router.post("/locate", response_model=LocateResponse)
async def locate_element(request: LocateRequest):
"""
本地元素定位接口
流程：
1. 启动 Playwright 浏览器
2. 如果 auto_login=True，执行自动登录
3. 访问目标页面
4. 提取可交互元素（含 micro-app 微前端）
5. 关键词算法匹配 + 置信度计算
6. 返回候选定位器列表
Args:
request: 定位请求参数
Returns:
LocateResponse: 定位结果
单个步骤元素定位
启动 Playwright → 登录 → 访问页面 → 提取元素 → 匹配定位器
"""
logger.info(f"开始元素定位: {request.step_description} @ {request.page_url}")
executor = None
try:
# 1. 创建执行器实例
# 1. 创建执行器实例
config = {
"headless": settings.PLAYWRIGHT_HEADLESS,
"timeout": 30000,
"screenshot": False, # 不需要执行步骤截图
"auto_login": request.auto_login,
}
config = {"headless": settings.PLAYWRIGHT_HEADLESS, "timeout": 30000, "screenshot": False, "auto_login": request.auto_login}
executor = PlaywrightExecutor(config)
# 2. 使用专用线程运行 Playwright（避免 asyncio 检测问题）
# 参考 execution_service.py 的已验证实现
result_container = {'success': False, 'candidates': [], 'message': '', 'screenshot': None}
def _playwright_worker():
"""Playwright 工作线程"""
try:
# 非 Main 线程需要绕过 asyncio 检测
import threading
import asyncio
if threading.current_thread() is not threading.main_thread():
import threading as _threading
if _threading.current_thread() is not _threading.main_thread():
asyncio.set_event_loop(None)
# 启动执行器
executor.start()
prep = _do_locate_with_executor(executor, request.page_url, request.auto_login)
if not prep['success']:
result_container['message'] = prep['message']
return
# 如果需要自动登录且未登录，执行登录
if request.auto_login and not executor._is_logged_in:
logger.info("执行自动登录...")
login_success = executor.do_login()
if not login_success:
result_container['message'] = "自动登录失败"
return
# 访问目标页面
logger.info(f"访问页面: {request.page_url}")
current_url = executor._page.url
target_url = request.page_url
# 智能跳转：如果已登录且目标 URL 是首页/登录页，不重复导航
is_login_page = 'login' in current_url.lower() or current_url.rstrip('/') == 'https://192.168.5.44'
is_target_login = target_url.rstrip('/') == 'https://192.168.5.44' or 'login' in target_url.lower()
if executor._is_logged_in and is_target_login and not is_login_page:
# 已登录且目标是登录页，直接在当前页面提取元素
logger.info(f"已在登录后页面，跳过导航。当前 URL: {current_url}")
else:
# 需要导航到目标页面
try:
executor._page.goto(target_url, wait_until="domcontentloaded", timeout=30000)
except Exception as e:
# networkidle 可能超时，改用 domcontentloaded
logger.warning(f"页面加载超时(networkidle)，尝试 domcontentloaded: {e}")
try:
executor._page.goto(target_url, wait_until="domcontentloaded", timeout=15000)
except Exception as e2:
logger.warning(f"页面加载超时(domcontentloaded)，继续使用当前页面: {e2}")
# 等待 SPA 渲染
executor._page.wait_for_timeout(3000)
# 提取可交互元素
logger.info("提取页面元素...")
elements = extract_interactive_elements(executor._page)
logger.info(f"提取到 {len(elements)} 个元素")
# 调试：打印前 10 个元素
for i, el in enumerate(elements[:10]):
logger.info(f" 元素{i}: tag={el.get('tag')}, type={el.get('type')}, placeholder={el.get('placeholder')}, text={el.get('text')}, id={el.get('id')}")
# 匹配元素
page_title = executor._page.title()
candidates = build_locator_candidates(
request.step_description,
elements,
page_title
)
candidates = build_locator_candidates(request.step_description, prep['elements'], prep['page_title'], request.page_url)
logger.info(f"匹配到 {len(candidates)} 个候选定位器")
# 截图（可选）
screenshot = None
try:
screenshot_bytes = executor._page.screenshot()
import base64
screenshot = base64.b64encode(screenshot_bytes).decode('utf-8')
screenshot = base64.b64encode(executor._page.screenshot()).decode('utf-8')
except Exception as e:
logger.warning(f"截图失败: {e}")
result_container['success'] = True
result_container['candidates'] = candidates
result_container['message'] = f"找到 {len(candidates)} 个候选定位器"
result_container['screenshot'] = screenshot
result_container.update({'success': True, 'candidates': candidates, 'message': f"找到 {len(candidates)} 个候选定位器", 'screenshot': screenshot})
except Exception as e:
logger.error(f"定位过程出错: {e}", exc_info=True)
result_container['success'] = False
result_container['message'] = f"定位失败: {str(e)}"
finally:
# 确保执行器停止
try:
executor.stop()
logger.info("执行器已停止")
except Exception as e:
logger.warning(f"停止执行器失败: {e}")
# 启动专用线程
worker_thread = threading.Thread(target=_playwright_worker, daemon=True)
worker_thread.start()
# 等待线程完成（最多 70 秒）
worker_thread.join(timeout=70)
if worker_thread.is_alive():
# 超时
logger.error("元素定位超时")
return LocateResponse(
success=False,
candidates=[],
message="定位超时，请检查页面是否可访问"
)
# 返回结果
return LocateResponse(success=False, candidates=[], message="定位超时")
return LocateResponse(**result_container)
except Exception as e:
logger.error(f"元素定位异常: {e}", exc_info=True)
return LocateResponse(
success=False,
candidates=[],
message=f"定位异常: {str(e)}"
)
\ No newline at end of file
return LocateResponse(success=False, candidates=[], message=f"定位异常: {str(e)}")
# ==================== 批量定位 API ====================
@router.post("/locate-batch", response_model=BatchLocateResponse)
async def locate_batch(request: BatchLocateRequest):
"""
批量元素定位（一键获取用例所有步骤的定位）
流程：
1. 从数据库获取用例的所有步骤
2. 启动 Playwright → 登录 → 访问页面
3. 提取页面元素
4. 为每个步骤匹配定位器
5. 更新用例步骤到数据库
6. 返回每个步骤的定位结果
"""
logger.info(f"开始批量定位: 用例 {request.case_id}")
# 延迟导入避免循环依赖
from app.services.case_service import CaseService
from app.database import async_session_maker
# 获取用例数据
async with async_session_maker() as session:
case_service = CaseService(session)
case = await case_service.get(request.case_id)
if not case:
return BatchLocateResponse(success=False, case_id=request.case_id, total_steps=0, located_steps=0, results=[], message="用例不存在")
steps = case.steps if isinstance(case.steps, list) else []
if not steps:
return BatchLocateResponse(success=False, case_id=request.case_id, total_steps=0, located_steps=0, results=[], message="用例没有步骤")
logger.info(f"用例有 {len(steps)} 个步骤")
# 在线程中执行 Playwright
result_container = {
'success': False, 'case_id': request.case_id,
'total_steps': len(steps), 'located_steps': 0,
'results': [], 'message': ''
}
updated_steps = [dict(s) if not isinstance(s, dict) else s for s in steps]
def _batch_worker():
executor = None
try:
import threading as _threading
if _threading.current_thread() is not _threading.main_thread():
asyncio.set_event_loop(None)
config = {"headless": settings.PLAYWRIGHT_HEADLESS, "timeout": 30000, "screenshot": False, "auto_login": request.auto_login}
executor = PlaywrightExecutor(config)
executor.start()
# 访问页面并提取元素（只做一次）
prep = _do_locate_with_executor(executor, request.page_url, request.auto_login)
if not prep['success']:
result_container['message'] = prep['message']
return
elements = prep['elements']
page_title = prep['page_title']
located_count = 0
# 为每个步骤匹配定位器
for step in updated_steps:
step_desc = step.get('name', '') or step.get('description', '')
step_order = step.get('order', 0)
if not step_desc.strip():
result_container['results'].append(StepLocateResult(
order=step_order, description=step_desc, success=False, message="步骤描述为空"
))
continue
# 跳过 navigate 类型的步骤（不需要定位器）
action = step.get('action', '')
if action == 'navigate':
result_container['results'].append(StepLocateResult(
order=step_order, description=step_desc, success=True,
locator_type="", locator_value="", confidence=1.0, message="导航步骤无需定位"
))
located_count += 1
continue
# 匹配定位器
candidates = build_locator_candidates(step_desc, elements, page_title, request.page_url)
if candidates:
best = candidates[0]
# 更新步骤的定位信息（使用驼峰命名匹配前端 IntentStepEditor）
step['locatorType'] = best['locator_type']
step['locatorValue'] = best['locator_value']
# 兼容蛇形命名（如果其他地方需要）
step['locator_type'] = best['locator_type']
step['locator_value'] = best['locator_value']
# 同步到 params.selector
if best['locator_value'] and 'params' in step:
step['params']['selector'] = best['locator_value']
elif best['locator_value']:
step['params'] = {'selector': best['locator_value']}
result_container['results'].append(StepLocateResult(
order=step_order, description=step_desc, success=True,
locator_type=best['locator_type'], locator_value=best['locator_value'],
confidence=best['confidence'], message=f"置信度 {best['confidence']:.0%}"
))
located_count += 1
else:
result_container['results'].append(StepLocateResult(
order=step_order, description=step_desc, success=False,
message="未找到匹配元素"
))
result_container['located_steps'] = located_count
result_container['success'] = True
result_container['message'] = f"成功定位 {located_count}/{len(steps)} 个步骤"
except Exception as e:
logger.error(f"批量定位出错: {e}", exc_info=True)
result_container['message'] = f"批量定位失败: {str(e)}"
finally:
if executor:
try:
executor.stop()
except:
pass
worker_thread = threading.Thread(target=_batch_worker, daemon=True)
worker_thread.start()
worker_thread.join(timeout=90)
if worker_thread.is_alive():
return BatchLocateResponse(success=False, case_id=request.case_id, total_steps=len(steps), located_steps=0, results=[], message="批量定位超时")
# 如果定位成功，更新数据库中的用例步骤
if result_container['success'] and result_container['located_steps'] > 0:
try:
async with async_session_maker() as session:
case_service = CaseService(session)
await case_service.update(request.case_id, {'steps': updated_steps})
logger.info(f"用例 {request.case_id} 步骤已更新到数据库")
except Exception as e:
logger.warning(f"更新数据库失败: {e}")
return BatchLocateResponse(**result_container)
frontend/src/views/Cases.vue
浏览文件 @ b2854e30
......@@ -176,7 +176,7 @@
{{ formatTime(row.created_at) }}
</template>
</el-table-column>
<el-table-column label="操作" width="200" fixed="right">
<el-table-column label="操作" width="280" fixed="right">
<template #default="{ row }">
<el-button size="small" type="primary" link @click="showEditDialog(row)">
编辑
......@@ -184,6 +184,16 @@
<el-button size="small" type="primary" link @click="viewCaseDetail(row)">
查看
</el-button>
<el-button
size="small"
type="success"
link
@click="batchLocate(row)"
:loading="locatingCaseId === row.id"
:disabled="row.caseType !== 'ui'"
>
获取定位
</el-button>
<el-button size="small" type="warning" link @click="copyCase(row)">
复制
</el-button>
......@@ -361,13 +371,36 @@
v-if="currentCase.steps && currentCase.steps.length > 0"
>
<el-table-column label="#" width="50" type="index" />
<el-table-column label="步骤名称" prop="name" min-width="150" />
<el-table-column label="动作" width="100">
<el-table-column label="步骤名称" prop="name" min-width="120" />
<el-table-column label="动作" width="90">
<template #default="{ row }">
<el-tag size="small" effect="plain">{{ row.action }}</el-tag>
</template>
</el-table-column>
<el-table-column label="参数" min-width="200">
<el-table-column label="定位类型" width="100">
<template #default="{ row }">
<el-tag
v-if="row.locatorType || row.locator_type"
size="small"
type="success"
>
{{ row.locatorType || row.locator_type }}
</el-tag>
<span v-else class="text-gray">-</span>
</template>
</el-table-column>
<el-table-column label="定位值" min-width="180">
<template #default="{ row }">
<code
v-if="row.locatorValue || row.locator_value || row.params?.selector"
class="locator-code"
>
{{ row.locatorValue || row.locator_value || row.params?.selector }}
</code>
<span v-else class="text-gray">-</span>
</template>
</el-table-column>
<el-table-column label="参数" min-width="150">
<template #default="{ row }">
<span class="params-text">{{ formatParams(row.params) }}</span>
</template>
......@@ -493,6 +526,7 @@ import {
import { caseApi } from '@/api/cases'
import { moduleApi, buildModuleTree } from '@/api/modules'
import { executionApi } from '@/api/executions'
import { elementLocateApi } from '@/api/elementLocate'
import CaseStepEditor from '@/components/CaseStepEditor.vue'
import IntentStepEditor from '@/components/IntentStepEditor.vue'
import { toStepDefinition, toIntentStep } from '@/utils/stepParser'
......@@ -689,7 +723,11 @@ const formatParams = (params: any) => {
if (!params) return '-'
if (typeof params === 'string') return params
try {
return JSON.stringify(params)
// 排除 selector（已在定位值列单独显示）
const { selector, ...rest } = params
const keys = Object.keys(rest)
if (keys.length === 0) return '-'
return JSON.stringify(rest)
} catch {
return '-'
}
......@@ -841,6 +879,57 @@ const deleteCase = async (row: any) => {
}
}
// ==================== 批量获取定位 ====================
/** 当前正在获取定位的用例 ID */
const locatingCaseId = ref('')
/**
* 一键获取用例所有步骤的元素定位
*
* 流程：
* 1. 调用批量定位 API
* 2. 后端启动 Playwright → 登录 → 访问页面 → 提取元素
* 3. 为每个步骤匹配定位器并更新到数据库
* 4. 前端显示结果
*/
const batchLocate = async (row: any) => {
try {
await ElMessageBox.confirm(
`将为用例「${row.name}」的所有步骤自动获取元素定位，是否继续？`,
'批量获取定位',
{ type: 'info', confirmButtonText: '开始获取', cancelButtonText: '取消' }
)
} catch {
return
}
locatingCaseId.value = row.id
try {
ElMessage.info('正在启动浏览器获取定位，请稍候...')
const result = await elementLocateApi.locateBatch({
case_id: row.id,
page_url: 'https://192.168.5.44/',
auto_login: true,
})
if (result.success) {
const msg = `定位完成：成功 ${result.located_steps}/${result.total_steps} 个步骤`
ElMessage.success(msg)
// 刷新用例列表
await loadCases()
} else {
ElMessage.warning(result.message || '定位失败')
}
} catch (error: any) {
ElMessage.error('批量定位失败: ' + (error.message || '未知错误'))
} finally {
locatingCaseId.value = ''
}
}
const runSelectedCases = async () => {
if (selectedCases.value.length === 0) return
......@@ -1032,6 +1121,19 @@ onMounted(() => {
color: #909399;
word-break: break-all;
}
.locator-code {
font-family: 'Consolas', 'Monaco', monospace;
font-size: 11px;
background: #f5f7fa;
padding: 2px 6px;
border-radius: 3px;
color: #409eff;
display: inline-block;
max-width: 100%;
overflow: hidden;
text-overflow: ellipsis;
}
}
}
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论