提交 0d1979ff authored 作者: 陈泽健's avatar 陈泽健

feat: UI用例录入优化 + 设备模拟模块 + 系统管理菜单

- UI用例录入:意图步骤编辑器、AI定位(宿主机 Playwright + Claude Code)
- 设备模拟:门口屏/无纸化/中控/集控客户端模拟 + MQTT 管理
- 系统管理:系统配置/信息/日志父菜单重构
- 修复 AI 定位 SSH 调用路径与 python3 命令
- 新增 .gitignore 排除缓存与调试产物
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 c6a688e4
# Python
__pycache__/
*.py[cod]
*.egg-info/
.venv/
venv/
*.egg
# Node / Frontend
node_modules/
frontend/dist/
frontend/dist-ssr/
*.local
# Data & runtime
data/
*.db
*.sqlite3
# Debug / temp
backend/debug_*.png
backend/scripts/debug_*.py
*.log
.DS_Store
Thumbs.db
# IDE / Editor
.idea/
.vscode/
*.swp
*.swo
# Env
.env
.env.*
!.env.example
# Coverage / test
.coverage
htmlcov/
.pytest_cache/
.mypy_cache/
# PRD:新增系统管理父菜单(系统配置 / 系统信息 / 操作日志)
> **版本**:v1.0
> **创建日期**:2026-07-29
> **状态**:待实现
> **关联项目**:平台自动化测试可视化系统 (platform-auto-test)
---
## 一、需求背景
当前侧边栏中「系统配置」为单个菜单项,路由 `/settings` 指向 `Settings.vue`(目前为占位页面)。随着平台功能增多,需要将系统级别的管理功能整合为统一的父菜单分组,方便后续扩展更多系统管理功能。
## 二、需求描述
### 2.1 侧边栏变更
将当前侧边栏中的单个「系统配置」菜单项,替换为 **「系统管理」父菜单**,展开后包含三个子菜单:
| 菜单项 | 路由 | 说明 |
|--------|------|------|
| 系统配置 | `/system/settings` | 原 `/settings` 功能迁移至此 |
| 系统信息 | `/system/info` | 新增:展示平台运行环境信息 |
| 操作日志 | `/system/logs` | 新增:展示系统操作日志记录 |
### 2.2 页面规划
#### 2.2.1 系统配置 (`/system/settings`)
- 将当前 `Settings.vue` 迁移至新路由
- 功能待后续 Phase 5 实现,当前保留占位状态
#### 2.2.2 系统信息 (`/system/info`) — 新增
展示平台运行环境信息,包括:
- 后端版本、Python 版本、运行时间
- 数据库状态(连接数、大小)
- 前端版本、Node 版本
- 部署环境信息
**数据来源**:后端新增 `/api/system/info` 接口
#### 2.2.3 操作日志 (`/system/logs`) — 新增
展示系统关键操作日志,包括:
- 用例执行记录列表(最近的执行记录)
- 系统配置变更记录(预留)
- 支持按时间范围、操作类型筛选
**数据来源**:复用现有执行记录 API `/api/executions`
### 2.3 不涉及变更
- 不影响其他路由(总览、模块管理、用例管理、录制、执行中心、报告中心、设备模拟)
- 不影响数据库表结构
- 不涉及后端模型变更
- 不涉及 Playwright 执行引擎
## 三、变更范围
| 层面 | 变更内容 | 影响范围 |
|------|---------|---------|
| **路由** | 新增 `/system` 父路由 + 3 个子路由 | `frontend/src/router/index.ts` |
| **侧边栏** | 替换单菜单为父菜单 | `frontend/src/App.vue` |
| **页面迁移** | Settings.vue 迁移至 `/system/settings` | 路由路径变更,组件内容不变 |
| **新增页面** | SystemInfo.vue | 新建文件 |
| **新增页面** | OperationLogs.vue | 新建文件 |
| **新增 API** | `/api/system/info` 返回系统信息 | `backend/app/routers/system.py` |
| **后端路由注册** | 注册 system 路由 | `backend/app/main.py` |
## 四、验收标准
1. ✅ 侧边栏显示「系统管理」父菜单,展开后含 3 个子菜单
2. ✅ 点击「系统配置」→ 跳转 `/system/settings`,显示原 Settings 页面内容
3. ✅ 点击「系统信息」→ 跳转 `/system/info`,显示运行环境信息
4. ✅ 点击「操作日志」→ 跳转 `/system/logs`,显示执行记录列表
5. ✅ 旧路由 `/settings` → 301 重定向到 `/system/settings`
6. ✅ 其他窗口开发的功能不受影响
7. ✅ 页面标题正确显示
## 五、注意事项
- 旧路由 `/settings` 需做重定向,避免书签/链接失效
- 设备模拟模块的 `/device-sim` 路由命名风格已为嵌套路由,本次参照其模式实现
- 当前 Settings.vue 为占位页面,仅迁移路由,不改动其内容
- 系统信息页面数据从后端获取,需确保后端接口返回正确的格式
\ No newline at end of file
# 执行计划:新增系统管理父菜单(系统配置 / 系统信息 / 操作日志)
> **版本**:v1.0
> **创建日期**:2026-07-29
> **来源**:`_PRD_需求文档_新增系统管理父菜单.md`
> **状态**:待执行
---
## 一、任务分解
| 阶段 | 任务 | 预计文件数 | 优先级 |
|------|------|-----------|--------|
| Phase 1 | 后端新增 `/api/system/info` 接口 | 2 文件 | P0 |
| Phase 2 | 前端路由改造:新增 `/system` 父路由 + 子路由 | 1 文件 | P0 |
| Phase 3 | 侧边栏改造:替换单菜单为父菜单 | 1 文件 | P0 |
| Phase 4 | 迁移 Settings.vue 到新路径 | 1 文件 | P0 |
| Phase 5 | 新建 SystemInfo.vue 页面 | 1 文件 | P0 |
| Phase 6 | 新建 OperationLogs.vue 页面 | 1 文件 | P0 |
| Phase 7 | 验证:启动服务 + 检查所有路由 | — | P1 |
---
## 二、详细实现步骤
### Phase 1: 后端新增 `/api/system/info` 接口
**文件**`backend/app/routers/system.py`(新建)
**接口定义**
```
GET /api/system/info → {
"app_name": "平台自动化测试可视化系统",
"backend_version": "2.0.0",
"python_version": "3.10.x",
"uptime": "运行时间(秒)",
"db_type": "mysql" | "sqlite",
"db_size": "数据库大小(MB)",
"total_modules": 28,
"total_cases": 283,
"frontend_version": "1.0.0"
}
```
**文件**`backend/app/main.py`(修改)
- 导入 `system` 路由
- 注册 `app.include_router(system.router)``app_include_router` 列表中
### Phase 2: 前端路由改造
**文件**`frontend/src/router/index.ts`(修改)
变更内容:
1. 删除旧的 `/settings` 路由
2. 新增 `/system` 父路由,包含 3 个子路由
```typescript
{
path: '/system',
name: 'System',
redirect: '/system/settings',
meta: { title: '系统管理' },
children: [
{
path: 'settings',
name: 'SystemSettings',
component: () => import('@/views/system/Settings.vue'),
meta: { title: '系统配置' }
},
{
path: 'info',
name: 'SystemInfo',
component: () => import('@/views/system/Info.vue'),
meta: { title: '系统信息' }
},
{
path: 'logs',
name: 'OperationLogs',
component: () => import('@/views/system/Logs.vue'),
meta: { title: '操作日志' }
}
]
}
```
3. 保留旧路由 `/settings` 作为重定向:
```typescript
{
path: '/settings',
redirect: '/system/settings'
}
```
### Phase 3: 侧边栏改造
**文件**`frontend/src/App.vue`(修改)
变更内容:
1. 删除原有的 `<el-menu-item index="/settings">` 菜单项
2. 新增 "系统管理" 父菜单(`<el-sub-menu>`):
```html
<el-sub-menu index="/system">
<template #title>
<el-icon><Tools /></el-icon>
<span>系统管理</span>
</template>
<el-menu-item index="/system/settings">系统配置</el-menu-item>
<el-menu-item index="/system/info">系统信息</el-menu-item>
<el-menu-item index="/system/logs">操作日志</el-menu-item>
</el-sub-menu>
```
3. 导入 `Tools` 图标(从 element-plus/icons-vue)
4. 更新 `currentRoute` 计算属性中的 `defaultType` 回退逻辑,添加 `/system` 支持
5. 更新 `pageTitle` 计算属性中的 `baseMap`,添加 `/system` 映射
### Phase 4: 迁移 Settings.vue
**文件**`frontend/src/views/Settings.vue``frontend/src/views/system/Settings.vue`(移动)
- 内容不变,保持当前占位状态
- 仅更新文件头部注释中的路径信息
### Phase 5: 新建 SystemInfo.vue
**文件**`frontend/src/views/system/Info.vue`(新建)
页面内容:
- 标题:系统信息
- 展示卡片:应用信息、运行环境、数据库状态
- 调用 `GET /api/system/info` 获取数据
- 使用 el-descriptions 或 el-card 展示
**文件**`frontend/src/api/system.ts`(新建)
```typescript
import request from '@/utils/request'
export function getSystemInfo() {
return request.get('/api/system/info')
}
```
### Phase 6: 新建 OperationLogs.vue
**文件**`frontend/src/views/system/Logs.vue`(新建)
页面内容:
- 标题:操作日志
- 展示最近执行记录列表
- 复用现有 `/api/executions` 接口
- 表格展示:执行时间、用例数、通过数、失败数、状态、操作人
- 支持按时间范围筛选
### Phase 7: 验证
| 验证项 | 预期结果 |
|--------|---------|
| 侧边栏显示「系统管理」父菜单 | 展开后含 3 个子菜单 |
| 点击「系统配置」 | 跳转 `/system/settings`,显示占位页面 |
| 点击「系统信息」 | 跳转 `/system/info`,显示系统信息 |
| 点击「操作日志」 | 跳转 `/system/logs`,显示执行记录 |
| 旧路由 `/settings` | 301 重定向到 `/system/settings` |
| 页面标题 | 各页面标题正确显示 |
| 其他菜单 | 不受影响(总览、模块管理、用例管理等) |
---
## 三、文件变更清单
### 新增文件
| # | 文件 | 说明 |
|---|------|------|
| 1 | `backend/app/routers/system.py` | 系统信息 API 路由 |
| 2 | `frontend/src/api/system.ts` | 系统信息 API 客户端 |
| 3 | `frontend/src/views/system/Info.vue` | 系统信息页面 |
| 4 | `frontend/src/views/system/Logs.vue` | 操作日志页面 |
| 5 | `Docs/PRD/需求文档/系统管理/_PRD_需求文档_新增系统管理父菜单.md` | PRD 需求文档 |
| 6 | `Docs/PRD/需求文档/系统管理/_PRD_需求文档_新增系统管理父菜单_计划执行.md` | 本执行计划文档 |
### 修改文件
| # | 文件 | 变更说明 |
|---|------|---------|
| 1 | `backend/app/main.py` | 注册 system 路由 |
| 2 | `frontend/src/router/index.ts` | 新增 `/system` 父路由 + 子路由,`/settings` 重定向 |
| 3 | `frontend/src/App.vue` | 侧边栏替换为系统管理父菜单 |
### 移动文件
| # | 文件 | 说明 |
|---|------|------|
| 1 | `frontend/src/views/Settings.vue``frontend/src/views/system/Settings.vue` | 迁移至新路径 |
---
## 四、多窗口开发注意事项
根据 `Docs/多窗口并行开发指南.md`,本次开发属于「纯代码开发」类型:
| 检查项 | 本次操作 |
|--------|---------|
| 后端端口 | 使用 8001(如已被占用则用 8002) |
| 数据库操作 | **不涉及**(无模型变更) |
| Playwright 执行 | **不涉及** |
| Git 冲突风险 | 低 — 修改的文件(router/index.ts / App.vue / main.py)与其他窗口交集小 |
| 提交前 Pull | 是,先拉取其他窗口的变更 |
---
## 五、回滚方案
若需回滚:
1. 路由:删除 `/system` 路由,恢复 `/settings` 路由
2. 侧边栏:恢复 `<el-menu-item index="/settings">`
3. 页面:Settings.vue 移回 `views/` 根目录
4. 后端:删除 `system` 路由注册
5. 删除新增文件
\ No newline at end of file
此差异已折叠。
...@@ -52,6 +52,12 @@ class Settings: ...@@ -52,6 +52,12 @@ class Settings:
SCREENSHOT_DIR: str = os.getenv("SCREENSHOT_DIR", "./data/screenshots") SCREENSHOT_DIR: str = os.getenv("SCREENSHOT_DIR", "./data/screenshots")
REPORT_DIR: str = os.getenv("REPORT_DIR", "./data/reports") REPORT_DIR: str = os.getenv("REPORT_DIR", "./data/reports")
# 设备模拟配置
DEVICE_SIM_MAX_INSTANCES: int = int(os.getenv("DEVICE_SIM_MAX_INSTANCES", "50"))
DEVICE_SIM_DEFAULT_REPORT_INTERVAL: int = int(os.getenv("DEVICE_SIM_DEFAULT_REPORT_INTERVAL", "30"))
DEVICE_SIM_LOG_RETENTION_DAYS: int = int(os.getenv("DEVICE_SIM_LOG_RETENTION_DAYS", "7"))
ENCRYPTION_KEY: str = os.getenv("ENCRYPTION_KEY", "")
# 日志配置 # 日志配置
LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO") LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
......
...@@ -120,6 +120,9 @@ async def _ensure_columns(conn) -> None: ...@@ -120,6 +120,9 @@ async def _ensure_columns(conn) -> None:
""" """
from sqlalchemy import text, inspect from sqlalchemy import text, inspect
# 导入模型以确保表被创建(必须导入才会注册到 Base.metadata)
from app.models import device_sim # noqa: F401
# 需要补齐的字段:[(表名, 列名, 列定义SQL)] # 需要补齐的字段:[(表名, 列名, 列定义SQL)]
columns_to_add = [ columns_to_add = [
("modules", "module_type", "VARCHAR(20) DEFAULT 'standard'"), ("modules", "module_type", "VARCHAR(20) DEFAULT 'standard'"),
......
...@@ -27,7 +27,7 @@ from fastapi.responses import FileResponse ...@@ -27,7 +27,7 @@ from fastapi.responses import FileResponse
from app.config import settings from app.config import settings
from app.database import init_db from app.database import init_db
from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies, security from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies, security, ai_locator, device_sim, system
# 配置日志 # 配置日志
logging.basicConfig( logging.basicConfig(
...@@ -146,6 +146,24 @@ app.include_router( ...@@ -146,6 +146,24 @@ app.include_router(
tags=["安全测试"] tags=["安全测试"]
) )
app.include_router(
ai_locator.router,
prefix="/api/ai",
tags=["AI定位"]
)
app.include_router(
device_sim.router,
prefix="/api/device-sim",
tags=["设备模拟"]
)
app.include_router(
system.router,
prefix="/api/system",
tags=["系统管理"]
)
# ==================== 根路径 ==================== # ==================== 根路径 ====================
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:device_sim.py
模块描述:设备模拟模块数据库模型定义
作者:czj
创建日期:2026-07-29
最后修改:2026-07-29
"""
from datetime import datetime
from typing import Optional
from sqlalchemy import String, Text, Integer, Boolean, DateTime, JSON
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base
class EnvConfig(Base):
"""
MQTT 环境配置数据库模型
管理 MQTT Broker 连接参数,是设备模拟模块的基础配置。
每个环境配置包含 Broker 地址、端口、认证信息、主题前缀等。
Attributes:
id (str): 配置唯一标识
name (str): 配置名称
broker_host (str): Broker 地址
broker_port (int): 端口号
username (str): MQTT 用户名
password_encrypted (str): 密码(加密存储)
topic_prefix (str): 主题前缀
use_tls (bool): 是否启用 TLS
client_id_prefix (str): 客户端 ID 前缀
status (str): 连接状态 (disconnected/connected/error)
last_connected_at (datetime): 最后连接时间
created_at (datetime): 创建时间
updated_at (datetime): 更新时间
"""
__tablename__ = "device_env_configs"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
name: Mapped[str] = mapped_column(String(100), nullable=False, comment="配置名称")
broker_host: Mapped[str] = mapped_column(String(255), nullable=False, comment="Broker 地址")
broker_port: Mapped[int] = mapped_column(Integer, default=1883, comment="端口号")
username: Mapped[Optional[str]] = mapped_column(String(100), nullable=True, comment="MQTT 用户名")
password_encrypted: Mapped[Optional[str]] = mapped_column(String(500), nullable=True, comment="密码(加密存储)")
topic_prefix: Mapped[str] = mapped_column(String(200), default="device", comment="主题前缀")
use_tls: Mapped[bool] = mapped_column(Boolean, default=False, comment="是否启用 TLS")
client_id_prefix: Mapped[str] = mapped_column(String(100), default="sim", comment="客户端 ID 前缀")
status: Mapped[str] = mapped_column(String(20), default="disconnected", comment="连接状态")
last_connected_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True, comment="最后连接时间")
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, comment="创建时间")
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间")
class DeviceSimulator(Base):
"""
模拟设备数据库模型
存储模拟设备的基本信息和运行状态。
支持 5 种设备类型:door(门口屏)、paperless(无纸化)、central(中控)、client(集控客户端)
Attributes:
id (str): 设备唯一标识
env_config_id (str): 关联环境配置 ID
device_type (str): 设备类型
device_name (str): 模拟设备名称
device_id (str): 模拟设备 ID(MQTT 主题中使用)
status (str): 运行状态 (stopped/running/error)
auto_reconnect (bool): 是否自动重连
metadata (dict): 设备自定义属性
last_reported_at (datetime): 最后上报时间
total_reports (int): 累计上报次数
created_at (datetime): 创建时间
updated_at (datetime): 更新时间
"""
__tablename__ = "device_simulators"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
env_config_id: Mapped[str] = mapped_column(String(64), nullable=False, comment="关联环境配置 ID")
device_type: Mapped[str] = mapped_column(String(50), nullable=False, comment="设备类型")
device_name: Mapped[str] = mapped_column(String(200), nullable=False, comment="模拟设备名称")
device_id: Mapped[str] = mapped_column(String(200), nullable=False, comment="模拟设备 ID")
status: Mapped[str] = mapped_column(String(20), default="stopped", comment="运行状态")
auto_reconnect: Mapped[bool] = mapped_column(Boolean, default=True, comment="是否自动重连")
extra_attrs: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True, comment="设备自定义属性")
last_reported_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True, comment="最后上报时间")
total_reports: Mapped[int] = mapped_column(Integer, default=0, comment="累计上报次数")
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, comment="创建时间")
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间")
class ReportLog(Base):
"""
上报记录数据库模型
记录设备上报和接收的消息历史。
Attributes:
id (str): 记录唯一标识
device_id (str): 关联模拟设备 ID
topic (str): MQTT 主题
payload (dict): 上报数据内容
direction (str): 方向 (publish/subscribe)
status (str): 发送状态 (success/failed)
error_message (str): 错误信息
created_at (datetime): 创建时间
"""
__tablename__ = "device_report_logs"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
device_id: Mapped[str] = mapped_column(String(64), nullable=False, comment="关联模拟设备 ID")
topic: Mapped[str] = mapped_column(String(500), nullable=False, comment="MQTT 主题")
payload: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True, comment="上报数据内容")
direction: Mapped[str] = mapped_column(String(10), default="publish", comment="方向")
status: Mapped[str] = mapped_column(String(20), default="success", comment="发送状态")
error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True, comment="错误信息")
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, comment="创建时间")
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:ai_locator.py
模块描述:AI 定位路由 — 通过宿主机 Claude Code 获取元素定位信息
功能:
1. POST /api/ai/locate — 创建定位任务,SSH 调用宿主机执行定位脚本
2. GET /api/ai/locate/{task_id}/result — 查询定位结果
工作流程:
1. 前端请求定位 → 后端创建任务文件 (data/ai_locate_tasks/{task_id}.json)
2. 后端通过 SSH 调用宿主机脚本:python scripts/ai_locate_worker.py --task-id TASK_ID
3. 宿主机脚本:Playwright 开网页 → 提取 DOM → Claude Code 分析匹配 → 写结果文件
4. 前端轮询获取结果
作者:czj
创建日期:2026-07-29
"""
import json
import logging
import os
import subprocess
import uuid
from pathlib import Path
from typing import Optional, List, Dict, Any
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
logger = logging.getLogger(__name__)
router = APIRouter()
# ==================== 目录配置 ====================
# 任务文件目录(与项目 data 目录一致)
TASK_DIR = Path(__file__).parent.parent.parent / "data" / "ai_locate_tasks"
# 部署目录(服务器上)
DEPLOY_DIR = "/data/third_party/plat-auto-test"
# 宿主机 SSH 配置
HOST_USER = "ubains"
HOST = "192.168.5.60"
# ==================== Schema 定义 ====================
class LocateRequest(BaseModel):
"""定位请求"""
step_description: str = Field(..., description="步骤描述,如「输入用户名」")
page_url: str = Field(..., description="目标网页 URL")
expected: str = Field(default="", description="预期结果(可选)")
class LocateCandidate(BaseModel):
"""定位候选"""
locator_type: str = Field(..., description="定位类型: css/xpath/text/id/name")
locator_value: str = Field(..., description="定位值")
confidence: float = Field(..., ge=0, le=1, description="置信度 0-1")
element_info: Dict[str, Any] = Field(default_factory=dict, description="元素信息")
class LocateTaskResponse(BaseModel):
"""创建任务响应"""
task_id: str = Field(..., description="任务ID")
status: str = Field(..., description="任务状态: pending/completed/failed")
message: str = Field(default="", description="提示信息")
class LocateResultResponse(BaseModel):
"""查询结果响应"""
task_id: str = Field(..., description="任务ID")
status: str = Field(..., description="任务状态: pending/completed/failed")
success: bool = Field(False, description="是否成功")
candidates: List[LocateCandidate] = Field(default_factory=list, description="定位候选列表")
message: str = Field(default="", description="提示信息")
# ==================== 路由 ====================
@router.post("/locate", response_model=LocateTaskResponse)
async def create_locate_task(data: LocateRequest):
"""
创建定位任务
1. 创建任务文件
2. 通过 SSH 调用宿主机 Claude Code 执行定位脚本
3. 返回 task_id,前端轮询结果
"""
try:
# 确保任务目录存在
TASK_DIR.mkdir(parents=True, exist_ok=True)
# 生成唯一任务 ID
task_id = f"locate_{uuid.uuid4().hex[:12]}"
# 创建任务文件
task_file = TASK_DIR / f"{task_id}.json"
task_data = {
"step_description": data.step_description,
"page_url": data.page_url,
"expected": data.expected,
}
with open(task_file, "w", encoding="utf-8") as f:
json.dump(task_data, f, ensure_ascii=False, indent=2)
logger.info(f"AI定位任务已创建: {task_id}, 步骤: {data.step_description}")
# SSH 调用宿主机执行定位脚本(异步,不阻塞)
ssh_cmd = (
f"ssh {HOST_USER}@{HOST} "
f"\"cd {DEPLOY_DIR}/backend && python3 scripts/ai_locate_worker.py --task-id {task_id}\""
)
try:
subprocess.Popen(
ssh_cmd,
shell=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
logger.info(f"已通过 SSH 调用宿主机定位脚本: {task_id}")
except Exception as e:
logger.warning(f"SSH 调用宿主机失败: {e},任务文件已创建,可手动执行")
# SSH 失败不影响任务创建,用户可以手动执行脚本
return LocateTaskResponse(
task_id=task_id,
status="pending",
message="定位任务已创建",
)
except Exception as e:
logger.error(f"创建定位任务失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"创建定位任务失败: {str(e)}")
@router.get("/locate/{task_id}/result", response_model=LocateResultResponse)
async def get_locate_result(task_id: str):
"""
查询定位结果
根据 task_id 读取结果文件,返回定位候选列表。
"""
try:
result_file = TASK_DIR / f"{task_id}_result.json"
if not result_file.exists():
# 结果文件不存在,检查任务文件是否存在
task_file = TASK_DIR / f"{task_id}.json"
if not task_file.exists():
return LocateResultResponse(
task_id=task_id,
status="failed",
success=False,
candidates=[],
message="任务不存在",
)
return LocateResultResponse(
task_id=task_id,
status="pending",
success=False,
candidates=[],
message="任务正在执行中",
)
# 读取结果文件
with open(result_file, "r", encoding="utf-8") as f:
result = json.load(f)
# 构建候选列表
candidates = []
for c in result.get("candidates", []):
candidates.append(LocateCandidate(
locator_type=c.get("locator_type", ""),
locator_value=c.get("locator_value", ""),
confidence=c.get("confidence", 0),
element_info=c.get("element_info", {}),
))
return LocateResultResponse(
task_id=task_id,
status=result.get("status", "completed"),
success=result.get("success", False),
candidates=candidates,
message=result.get("message", ""),
)
except Exception as e:
logger.error(f"查询定位结果失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"查询定位结果失败: {str(e)}")
\ No newline at end of file
此差异已折叠。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:system.py
模块描述:系统管理 API 路由(系统信息、运行环境)
作者:czj
创建日期:2026-07-29
最后修改:2026-07-29
"""
import logging
import platform
import sys
import time
from pathlib import Path
from fastapi import APIRouter, Depends
from sqlalchemy import func, select, text
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database import get_db
from app.models.module import Module
from app.models.test_case import TestCase
logger = logging.getLogger(__name__)
router = APIRouter()
# 应用启动时间(模块加载时记录)
_START_TIME = time.time()
@router.get("/info", response_model=dict)
async def get_system_info(db: AsyncSession = Depends(get_db)):
"""
获取系统运行信息
返回应用版本、运行环境、数据库状态、用例/模块统计等。
Returns:
dict: 系统信息
"""
# 数据库类型
is_sqlite = "sqlite" in settings.DATABASE_URL
db_type = "sqlite" if is_sqlite else "mysql"
# 数据库大小
db_size_mb = 0.0
try:
if is_sqlite:
# SQLite:直接读取文件大小
db_path = settings.DATABASE_URL.replace("sqlite+aiosqlite:///", "")
# 处理相对路径
p = Path(db_path)
if not p.is_absolute():
# 相对路径通常相对 backend/ 目录
p = Path.cwd() / p
if p.exists():
db_size_mb = round(p.stat().st_size / (1024 * 1024), 2)
else:
# MySQL:查询 information_schema
result = await db.execute(
text(
"SELECT ROUND(SUM(data_length + index_length) / 1024 / 1024, 2) "
"AS size_mb FROM information_schema.tables "
"WHERE table_schema = DATABASE()"
)
)
row = result.scalar()
db_size_mb = float(row) if row else 0.0
except Exception as e:
logger.warning(f"获取数据库大小失败: {e}")
db_size_mb = 0.0
# 模块 / 用例总数
total_modules = 0
total_cases = 0
try:
result = await db.execute(select(func.count(Module.id)))
total_modules = result.scalar() or 0
result = await db.execute(select(func.count(TestCase.id)))
total_cases = result.scalar() or 0
except Exception as e:
logger.warning(f"获取模块/用例统计失败: {e}")
# 运行时间
uptime_seconds = int(time.time() - _START_TIME)
hours, remainder = divmod(uptime_seconds, 3600)
minutes, seconds = divmod(remainder, 60)
uptime_text = f"{hours}小时{minutes}分钟{seconds}秒" if hours else f"{minutes}分钟{seconds}秒"
return {
"app_name": settings.APP_NAME,
"backend_version": settings.APP_VERSION,
"python_version": platform.python_version(),
"platform": platform.platform(),
"os": f"{platform.system()} {platform.release()}",
"uptime_seconds": uptime_seconds,
"uptime_text": uptime_text,
"db_type": db_type,
"db_size_mb": db_size_mb,
"total_modules": total_modules,
"total_cases": total_cases,
"debug": settings.DEBUG,
"playwright_headless": settings.PLAYWRIGHT_HEADLESS,
"frontend_version": "1.0.0",
}
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:device_sim.py
模块描述:设备模拟模块 Pydantic 模式定义
作者:czj
创建日期:2026-07-29
最后修改:2026-07-29
"""
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, Field, ConfigDict
# ==================== 环境配置 Schema ====================
class EnvConfigBase(BaseModel):
"""环境配置基础模式"""
name: str = Field(..., min_length=1, max_length=100, description="配置名称")
broker_host: str = Field(..., min_length=1, max_length=255, description="Broker 地址")
broker_port: int = Field(default=1883, ge=1, le=65535, description="端口号")
username: Optional[str] = Field(default=None, max_length=100, description="MQTT 用户名")
password: Optional[str] = Field(default=None, max_length=500, description="MQTT 密码")
topic_prefix: str = Field(default="device", max_length=200, description="主题前缀")
use_tls: bool = Field(default=False, description="是否启用 TLS")
client_id_prefix: str = Field(default="sim", max_length=100, description="客户端 ID 前缀")
class EnvConfigCreate(EnvConfigBase):
"""创建环境配置请求"""
pass
class EnvConfigUpdate(BaseModel):
"""更新环境配置请求"""
name: Optional[str] = Field(default=None, min_length=1, max_length=100, description="配置名称")
broker_host: Optional[str] = Field(default=None, min_length=1, max_length=255, description="Broker 地址")
broker_port: Optional[int] = Field(default=None, ge=1, le=65535, description="端口号")
username: Optional[str] = Field(default=None, max_length=100, description="MQTT 用户名")
password: Optional[str] = Field(default=None, max_length=500, description="MQTT 密码")
topic_prefix: Optional[str] = Field(default=None, max_length=200, description="主题前缀")
use_tls: Optional[bool] = Field(default=None, description="是否启用 TLS")
client_id_prefix: Optional[str] = Field(default=None, max_length=100, description="客户端 ID 前缀")
class EnvConfigResponse(EnvConfigBase):
"""环境配置响应"""
id: str = Field(..., description="配置 ID")
status: str = Field(..., description="连接状态")
last_connected_at: Optional[datetime] = Field(default=None, description="最后连接时间")
created_at: datetime = Field(..., description="创建时间")
updated_at: datetime = Field(..., description="更新时间")
model_config = ConfigDict(from_attributes=True)
class EnvConfigListResponse(BaseModel):
"""环境配置列表响应"""
items: List[EnvConfigResponse] = Field(default=[], description="环境配置列表")
total: int = Field(default=0, description="总数")
# ==================== 模拟设备 Schema ====================
class SimulatorBase(BaseModel):
"""模拟设备基础模式"""
env_config_id: str = Field(..., description="关联环境配置 ID")
device_type: str = Field(..., pattern="^(door|paperless|central|client)$", description="设备类型")
device_name: str = Field(..., min_length=1, max_length=200, description="模拟设备名称")
device_id: str = Field(..., min_length=1, max_length=200, description="模拟设备 ID")
auto_reconnect: bool = Field(default=True, description="是否自动重连")
metadata: Optional[dict] = Field(default=None, description="设备自定义属性")
class SimulatorCreate(SimulatorBase):
"""创建模拟设备请求"""
pass
class SimulatorUpdate(BaseModel):
"""更新模拟设备请求"""
device_name: Optional[str] = Field(default=None, min_length=1, max_length=200, description="模拟设备名称")
auto_reconnect: Optional[bool] = Field(default=None, description="是否自动重连")
metadata: Optional[dict] = Field(default=None, description="设备自定义属性")
class SimulatorResponse(SimulatorBase):
"""模拟设备响应"""
id: str = Field(..., description="设备 ID")
status: str = Field(..., description="运行状态")
last_reported_at: Optional[datetime] = Field(default=None, description="最后上报时间")
total_reports: int = Field(default=0, description="累计上报次数")
created_at: datetime = Field(..., description="创建时间")
updated_at: datetime = Field(..., description="更新时间")
model_config = ConfigDict(from_attributes=True)
class SimulatorListResponse(BaseModel):
"""模拟设备列表响应"""
items: List[SimulatorResponse] = Field(default=[], description="模拟设备列表")
total: int = Field(default=0, description="总数")
# ==================== 上报记录 Schema ====================
class ReportLogResponse(BaseModel):
"""上报记录响应"""
id: str = Field(..., description="记录 ID")
device_id: str = Field(..., description="关联设备 ID")
topic: str = Field(..., description="MQTT 主题")
payload: Optional[dict] = Field(default=None, description="上报数据")
direction: str = Field(..., description="方向")
status: str = Field(..., description="状态")
error_message: Optional[str] = Field(default=None, description="错误信息")
created_at: datetime = Field(..., description="创建时间")
model_config = ConfigDict(from_attributes=True)
class ReportLogListResponse(BaseModel):
"""上报记录列表响应"""
items: List[ReportLogResponse] = Field(default=[], description="上报记录列表")
total: int = Field(default=0, description="总数")
# ==================== 手动上报 Schema ====================
class ManualReportRequest(BaseModel):
"""手动上报请求"""
topic: str = Field(..., min_length=1, max_length=500, description="上报主题")
payload: dict = Field(default={}, description="上报数据内容")
# ==================== 连接测试 Schema ====================
class TestConnectionResponse(BaseModel):
"""连接测试响应"""
success: bool = Field(..., description="是否成功")
message: str = Field(default="", description="结果信息")
latency_ms: Optional[int] = Field(default=None, description="连接延迟(毫秒)")
\ No newline at end of file
...@@ -32,11 +32,14 @@ class StepDefinition(BaseModel): ...@@ -32,11 +32,14 @@ class StepDefinition(BaseModel):
action (str): 动作类型 (navigate/click/fill/assert等) action (str): 动作类型 (navigate/click/fill/assert等)
params (dict): 动作参数 params (dict): 动作参数
expected (str): 预期结果 expected (str): 预期结果
actual (str): 实际结果(预留)
selectors (Optional[List[str]]): 多选择器回退链 selectors (Optional[List[str]]): 多选择器回退链
page_key (Optional[str]): SelectorMapper 页面标识 page_key (Optional[str]): SelectorMapper 页面标识
element_key (Optional[str]): 元素标识 element_key (Optional[str]): 元素标识
force (Optional[bool]): 强制点击 force (Optional[bool]): 强制点击
wait_after (Optional[int]): 自定义操作后延时(ms) wait_after (Optional[int]): 自定义操作后延时(ms)
locator_type (Optional[str]): 定位类型: css/xpath/text/id/name
locator_value (Optional[str]): 定位值
""" """
order: int = Field(..., ge=1, description="步骤顺序") order: int = Field(..., ge=1, description="步骤顺序")
...@@ -44,11 +47,14 @@ class StepDefinition(BaseModel): ...@@ -44,11 +47,14 @@ class StepDefinition(BaseModel):
action: str = Field(..., description="动作类型") action: str = Field(..., description="动作类型")
params: Dict[str, Any] = Field(default_factory=dict, description="动作参数") params: Dict[str, Any] = Field(default_factory=dict, description="动作参数")
expected: str = Field(default="", description="预期结果") expected: str = Field(default="", description="预期结果")
actual: str = Field(default="", description="实际结果")
selectors: Optional[List[str]] = Field(None, description="多选择器回退链") selectors: Optional[List[str]] = Field(None, description="多选择器回退链")
page_key: Optional[str] = Field(None, description="SelectorMapper 页面标识") page_key: Optional[str] = Field(None, description="SelectorMapper 页面标识")
element_key: Optional[str] = Field(None, description="元素标识") element_key: Optional[str] = Field(None, description="元素标识")
force: Optional[bool] = Field(None, description="强制点击") force: Optional[bool] = Field(None, description="强制点击")
wait_after: Optional[int] = Field(None, description="自定义操作后延时(ms)") wait_after: Optional[int] = Field(None, description="自定义操作后延时(ms)")
locator_type: Optional[str] = Field(None, description="定位类型: css/xpath/text/id/name")
locator_value: Optional[str] = Field(None, description="定位值")
class TestCaseBase(BaseModel): class TestCaseBase(BaseModel):
......
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
...@@ -8,4 +8,5 @@ pydantic==2.5.3 ...@@ -8,4 +8,5 @@ pydantic==2.5.3
python-multipart==0.0.6 python-multipart==0.0.6
playwright==1.40.0 playwright==1.40.0
websockets==12.0 websockets==12.0
requests==2.31.0 requests==2.31.0
\ No newline at end of file paho-mqtt==2.1.0
\ No newline at end of file
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
/**
* 系统管理 API 接口
*
* @module api/system
* @author czj
* @date 2026-07-29
*/
import request from '@/utils/request'
/** 系统信息响应结构 */
export interface SystemInfo {
app_name: string
backend_version: string
python_version: string
platform: string
os: string
uptime_seconds: number
uptime_text: string
db_type: string
db_size_mb: number
total_modules: number
total_cases: number
debug: boolean
playwright_headless: boolean
frontend_version: string
}
/**
* 系统管理 API 服务
*/
export const systemApi = {
/**
* 获取系统运行信息
*/
async getInfo(): Promise<SystemInfo> {
const response = await request.get('/api/system/info')
return response.data
},
}
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论