提交 5848a57b authored 作者: 陈泽健's avatar 陈泽健

feat: 本地元素定位功能 - 实际访问页面获取真实元素定位器

新增本地元素定位 API,通过 Playwright 实际访问被测页面,提取真实元素定位器:
- backend/app/routers/element_locator.py: 新增 /api/element/locate 端点,支持 auto_login 和 micro-app 微前端
- backend/app/main.py: 注册 element_locator 路由
- frontend/src/api/elementLocate.ts: 新增前端元素定位 API 接口
- frontend/src/components/IntentStepEditor.vue: 修改 handleLocate 调用本地 API,简化流程

优势:本地开发可用,无需容器化部署;响应更快;支持离线使用
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 b0f0d211
......@@ -27,7 +27,7 @@ from fastapi.responses import FileResponse
from app.config import settings
from app.database import init_db
from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies, security, ai_locator, device_sim, system
from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies, security, ai_locator, device_sim, system, element_locator
# 配置日志
logging.basicConfig(
......@@ -164,6 +164,12 @@ app.include_router(
tags=["系统管理"]
)
app.include_router(
element_locator.router,
prefix="/api/element",
tags=["元素定位"]
)
# ==================== 根路径 ====================
......
"""
元素定位路由
提供本地元素定位功能,通过 Playwright 访问被测页面,
提取可交互元素,匹配步骤描述,返回候选定位器列表。
@author czj
@date 2026-08-03
"""
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
import logging
import asyncio
import threading
from app.executors.playwright_executor import PlaywrightExecutor
from app.config import settings
logger = logging.getLogger(__name__)
router = APIRouter(tags=["元素定位"])
# ==================== Schema 定义 ====================
class LocateRequest(BaseModel):
"""元素定位请求"""
step_description: str # 步骤描述,如"输入用户名"
page_url: str # 目标页面 URL
expected: str = "" # 预期结果(可选)
auto_login: bool = True # 是否自动登录(默认 True)
class LocateCandidate(BaseModel):
"""定位候选"""
locator_type: str # css/xpath/text/id/name
locator_value: str # 定位值
confidence: float # 置信度 0-1
element_info: Dict[str, Any] # 元素信息
class LocateResponse(BaseModel):
"""元素定位响应"""
success: bool
candidates: List[LocateCandidate]
message: str
screenshot: Optional[str] = None # 页面截图(base64)
# ==================== 元素提取函数 ====================
def extract_interactive_elements(page) -> list:
"""
提取页面中所有可交互元素的信息
支持 micro-app 微前端容器
Args:
page: Playwright Page 对象
Returns:
list: 元素信息列表
"""
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 || '',
});
});
// 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 || '',
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 || '',
inIframe: true,
});
});
}
} catch (e) {
// 跨域 iframe 无法访问,忽略
}
});
return elements;
}''')
def extract_keywords(text: str) -> list:
"""
从步骤描述中提取关键词
策略:
1. 先去除动作词(输入/点击/选择等),保留目标词
2. 将剩余文本按语义拆分为短词
3. 同时返回原始词和拆分后的子词
Args:
text: 步骤描述文本
Returns:
list: 关键词列表
"""
# 去除动作词
action_words = ['输入', '填写', '填入', '键入', '写入', '点击', '按下', '选择',
'单击', '双击', '打开', '导航', '访问', '进入', '跳转', '前往',
'等待', '延时', '断言', '验证', '检查', '确认', '勾选', '取消',
'悬停', '滚动', '截图', '按钮', '链接', '输入框', '下拉框']
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.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 unique_keywords
def build_locator_candidates(step_description: str, elements: list, page_title: str = "") -> list:
"""
关键词算法匹配元素
返回候选定位器列表(按置信度降序)
Args:
step_description: 步骤描述
elements: 元素信息列表
page_title: 页面标题(用于上下文)
Returns:
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}")
for el in elements:
score = 0.0
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 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
# 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,
'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', ''),
}
})
# 按置信度降序排序
candidates.sort(key=lambda x: x['confidence'], reverse=True)
return candidates
# ==================== 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: 定位结果
"""
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,
}
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():
asyncio.set_event_loop(None)
# 启动执行器
executor.start()
# 如果需要自动登录且未登录,执行登录
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
)
logger.info(f"匹配到 {len(candidates)} 个候选定位器")
# 截图(可选)
screenshot = None
try:
screenshot_bytes = executor._page.screenshot()
import base64
screenshot = base64.b64encode(screenshot_bytes).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
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(**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
/**
* 元素定位 API
*
* 提供本地元素定位功能,通过 Playwright 访问被测页面,
* 提取可交互元素,匹配步骤描述,返回候选定位器列表。
*
* @module api/elementLocate
* @author czj
* @date 2026-08-03
*/
import request from '@/utils/request'
/**
* 元素定位请求参数
*/
export interface LocateRequest {
/** 步骤描述,如"输入用户名" */
step_description: string
/** 目标页面 URL */
page_url: string
/** 预期结果(可选) */
expected?: string
/** 是否自动登录(默认 true) */
auto_login?: boolean
}
/**
* 定位候选
*/
export interface LocateCandidate {
/** 定位类型: css/xpath/text/id/name */
locator_type: string
/** 定位值 */
locator_value: string
/** 置信度 0-1 */
confidence: number
/** 元素信息 */
element_info: Record<string, any>
}
/**
* 元素定位响应
*/
export interface LocateResponse {
/** 是否成功 */
success: boolean
/** 候选定位器列表(按置信度降序) */
candidates: LocateCandidate[]
/** 消息 */
message: string
/** 页面截图(base64,可选) */
screenshot?: string
}
/**
* 元素定位 API
*/
export const elementLocateApi = {
/**
* 执行元素定位
*
* @param data - 定位请求参数
* @returns 定位结果
*
* @example
* const result = await elementLocateApi.locate({
* step_description: '输入用户名',
* page_url: 'https://192.168.5.44/',
* auto_login: true,
* })
*
* if (result.success && result.candidates.length > 0) {
* const best = result.candidates[0]
* console.log(`定位成功: ${best.locator_value} (${best.confidence})`)
* }
*/
async locate(data: LocateRequest): Promise<LocateResponse> {
const response = await request.post('/api/element/locate', data, {
timeout: 60000, // 60秒超时(Playwright 启动较慢)
})
return response.data
},
}
\ No newline at end of file
......@@ -172,7 +172,7 @@ import { ref, computed } from 'vue'
import { ElMessage } from 'element-plus'
import { Top, Bottom, Delete, Plus, Search } from '@element-plus/icons-vue'
import type { IntentStep } from '@/utils/stepParser'
import { aiLocateApi } from '@/api/aiLocate'
import { elementLocateApi } from '@/api/elementLocate'
import { usePageUrl } from '@/utils/usePageUrl'
const props = defineProps<{
......@@ -238,15 +238,20 @@ const emitChange = () => {
emit('update:modelValue', steps.value)
}
// ==================== AI 定位 ====================
// ==================== 元素定位 ====================
/**
* 获取定位信息
* 获取定位信息(本地 Playwright 方式)
*
* 步骤:
* 1. 创建定位任务(后端写入任务文件)
* 2. 后端 SSH 调用宿主机 Claude Code 执行定位脚本
* 3. 轮询结果,自动填充到定位字段
* 流程:
* 1. 调用本地元素定位 API(/api/element/locate)
* 2. 后端启动 Playwright 访问页面、提取元素、匹配定位器
* 3. 自动填充定位信息到步骤中
*
* 优势:
* - 本地开发可用,无需容器化部署
* - 响应更快(无 SSH 开销)
* - 支持离线使用(不依赖 Claude Code)
*/
const handleLocate = async (index: number) => {
const step = steps.value[index]
......@@ -262,60 +267,33 @@ const handleLocate = async (index: number) => {
}
locatingIndex.value = index
try {
// 1. 创建定位任务
const task = await aiLocateApi.locate({
ElMessage.info('正在启动浏览器访问页面...')
// 调用本地元素定位 API
const result = await elementLocateApi.locate({
step_description: step.description,
page_url: pageUrl.value,
expected: step.expected,
auto_login: true, // 默认自动登录
})
ElMessage.info(`定位任务已创建 (${task.task_id}),正在等待结果...`)
// 2. 轮询结果(每 3 秒一次,最多 60 秒)
let retries = 20
while (retries > 0) {
await new Promise(r => setTimeout(r, 3000))
const result = await aiLocateApi.getResult(task.task_id)
if (result.status === 'completed') {
if (result.success && result.candidates.length > 0) {
// 取置信度最高的候选
const best = result.candidates[0]
step.locatorType = best.locator_type
step.locatorValue = best.locator_value
ElMessage.success(`定位成功 (置信度: ${Math.round(best.confidence * 100)}%)`)
} else {
ElMessage.warning(result.message || '未找到匹配元素,请手动输入定位信息')
}
emitChange()
return
}
if (result.status === 'failed') {
ElMessage.error('定位失败: ' + result.message)
return
}
retries--
}
// 超时:检查是否有结果文件但状态未更新
try {
const result = await aiLocateApi.getResult(task.task_id)
if (result.status === 'completed' && result.candidates?.length > 0) {
const best = result.candidates[0]
// 自动填充定位信息
step.locatorType = best.locator_type
step.locatorValue = best.locator_value
ElMessage.success(`定位成功 (置信度: ${Math.round(best.confidence * 100)}%)`)
ElMessage.success(`定位成功!置信度 ${(best.confidence * 100).toFixed(0)}%`)
emitChange()
return
}
} catch {
// 忽略
} else {
ElMessage.warning(result.message || '未找到匹配元素,请手动输入定位信息')
}
ElMessage.warning('定位超时,请确认宿主机上定位脚本已运行,或手动输入定位信息')
} catch (error: any) {
console.error('元素定位失败:', error)
ElMessage.error('定位失败: ' + (error.message || '未知错误'))
} finally {
locatingIndex.value = -1
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论