Skip to content

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

feat(ApiSecurityTest): 对接ERP系统实现安全测试任务创建功能 

- 添加ERP对接配置项包括API密钥、基础URL和重试机制
- 实现获取任务类型、状态、人员和需求列表的基础数据功能
- 集成创建任务接口支持设置任务基本信息和关联需求
- 添加上传截图功能用于任务描述中的富文本内容
- 实现基础数据缓存机制避免重复请求提高性能
- 集成日志记录功能追踪API调用和任务创建状态
- 更新.gitignore添加API安全测试日志忽略规则
- 创建ERP对接需求文档和技术接口文档
上级 46442e56
显示空白字符变更
内嵌 并排
正在显示 包含 4173 行增加 和 7 行删除
.gitignore
浏览文件 @ 387ab3eb
......@@ -15,6 +15,10 @@
# 新统一平台图片日志
/新统一平台/log/imgs/
# API安全测试日志
/AuxiliaryTool/ScriptTool/ApiSecurityTest/logs/
/AuxiliaryTool/ScriptTool/ApiSecurityTest/*.log
# --- Python 核心缓存与编译文件 (必须配置) ---
# 忽略所有 __pycache__ 目录 (包含任意层级的子目录)
......
AuxiliaryTool/ScriptTool/ApiSecurityTest/ERP_INTEGRATION_README.md 0 → 100644
浏览文件 @ 387ab3eb
# ERP对接任务创建功能说明
## 功能概述
本功能在现有接口安全测试系统基础上增加了ERP任务创建功能，实现测试完成后自动创建ERP任务进行跟踪管理。
## 业务流程
```
步骤1：执行API安全测试（已有）
步骤2：生成安全测试报告（已有）
步骤3：上传报告到网盘（已有）
步骤4：创建ERP任务（🆕 新功能）
```
## 配置说明
在 `config.yaml` 中新增了 `erp` 配置节点：
```yaml
# ERP对接配置
erp:
# 是否启用ERP对接
enabled: true
# ERP API配置
base_url: "https://office.ubainsyun.com:5082/api/uerp"
api_key: "your_api_key_here" # 需要替换为实际密钥
timeout: 30
retry_interval: 2
max_retries: 3
# 任务配置
task:
type_id: 2 # 任务类型：2-测试任务
status_id: 1 # 任务状态：1-新建
level: 3 # 紧急程度：3-紧急处理
excepthour: 8 # 期望工时：8小时
user_id: 20 # 责任人ID
createuser_id: 5 # 创建人ID
deadline_days: 3 # 截止天数：当前时间+3天
# 需求关联
develop_search_str: "【新统一平台】公司内部安全测试扫描及修复验证"
# 缓存配置
cache_duration: 3600 # 基础数据缓存时长（秒）
# 上传配置
upload:
enabled: true # 是否上传截图
max_images: 5 # 最多上传图片数量
# 网盘配置
nas:
base_path: "\\\\192.168.9.9\\deploy\\18其它系统\\安全测试\\04新统一平台-安全报告"
```
## 使用方式
### 1. 启用ERP对接
在 `config.yaml` 中设置：
```yaml
erp:
enabled: true
api_key: "your_actual_api_key" # 替换为实际的API密钥
```
### 2. 配置任务参数
根据实际情况调整：
- `user_id`: 责任人ID（从ERP人员列表中选择）
- `createuser_id`: 创建人ID（从ERP人员列表中选择）
- `level`: 默认紧急程度（1-5）
- `excepthour`: 期望工时（小时）
- `deadline_days`: 截止天数
### 3. 执行测试
正常执行安全测试即可：
```bash
# 执行全部测试
python run_all.py
# 执行单个模块
python run_all.py api01
```
### 4. 查看结果
测试完成后会自动创建ERP任务，控制台输出：
```
==========================================================================
ERP任务创建阶段
==========================================================================
步骤 1/5: 解析测试报告...
✓ 测试目标: https://192.168.5.44
✓ 测试时间: 2026-06-24 15:00:00 ~ 15:30:00
✓ 漏洞统计: 高危=3, 中危=7, 低危=4
步骤 2/5: 获取ERP基础数据...
✓ 任务类型: 10种
✓ 任务状态: 5种
✓ 人员列表: 45人
步骤 3/5: 搜索关联需求...
✓ 找到需求: 【新统一平台】公司内部安全测试扫描及修复验证 (ID: 128)
步骤 4/5: 构建任务数据...
任务数据:
- 名称: 192.168.5.44接口安全测试及漏洞修复_20260624
- 类型: 测试任务
- 状态: 新建
- 紧急程度: 5
- 期望工时: 8小时
- 截止时间: 2026-06-27 18:00:00
- 责任人: 郑晓兵
- 创建人: 陈泽键
- 关联需求: 【新统一平台】公司内部安全测试扫描及修复验证
步骤 5/5: 创建ERP任务...
==========================================================================
✓ ERP任务创建成功！
- 任务ID: 1024
- 任务名称: 192.168.5.44接口安全测试及漏洞修复_20260624
- 任务链接: https://office.ubainsyun.com:5082/task/detail/1024
==========================================================================
```
## 新增模块
### 1. ERP客户端 (`utils/erp_client.py`)
提供ERP API调用功能：
- 获取任务类型、状态、人员列表
- 搜索需求列表
- 创建任务
- 上传图片
- 自动重试机制
- 数据缓存
### 2. 报告解析器 (`utils/report_parser.py`)
从Markdown报告中提取信息：
- 测试目标URL
- 测试时间
- 漏洞统计
- 漏洞列表
- 生成任务名称
- 生成任务描述HTML
### 3. 任务创建器 (`utils/task_creator.py`)
协调完成任务创建：
- 解析报告
- 获取ERP基础数据
- 搜索关联需求
- 构建任务数据
- 调用ERP接口
- 日志记录
## 任务名称规则
格式：`{目标IP}接口安全测试及漏洞修复_{日期}`
示例：`192.168.5.44接口安全测试及漏洞修复_20260624`
## 紧急程度规则
根据漏洞数量自动计算：
- **5** - 有高危漏洞（最紧急）
- **4** - 有中危漏洞
- **3** - 有低危漏洞
- **2** - 仅有信息类
- **1** - 无漏洞
## 错误处理
- ERP创建失败不影响测试报告生成和上传
- 详细的错误日志便于排查
- 网络超时自动重试（最多3次）
- 配置错误时有明确提示
## 注意事项
1. **API密钥**：需要在配置中提供正确的ERP API密钥
2. **人员ID**：user_id 和 createuser_id 需要从ERP系统中获取
3. **需求搜索**：develop_search_str 需要准确匹配ERP中的需求名称
4. **网络连接**：确保能够访问ERP服务器
5. **权限验证**：API密钥需要有创建任务的权限
## 禁用ERP对接
如果不需要ERP对接功能，在配置中设置：
```yaml
erp:
enabled: false
```
测试流程将正常执行，但不会创建ERP任务。
AuxiliaryTool/ScriptTool/ApiSecurityTest/config.yaml
浏览文件 @ 387ab3eb
......@@ -64,6 +64,15 @@ report:
# 报告文件名前缀
prefix: "api_security_report"
# 日志配置
log:
# 日志输出目录
output_dir: "logs/"
# 是否启用文件日志记录
file_enabled: true
# 日志级别：DEBUG, INFO, WARNING, ERROR, CRITICAL
level: "INFO"
# Nginx 配置分析结果（来自 unified443.conf 分析）
nginx_analysis:
# 已知端口映射（从Nginx配置中提取的内部服务端口）
......@@ -123,3 +132,53 @@ limits:
batch_max: 50
# 请求间隔（秒），避免过快触发限流
request_interval: 0.5
# ERP对接配置
erp:
# 是否启用ERP对接
enabled: true
# ERP API基础URL
base_url: "https://office.ubainsyun.com:5082/api/uerp"
# ERP API密钥（需要替换为实际密钥）
api_key: "9adc1ce611544fae9bbbc5f6b1d6ce87"
# 请求超时时间（秒）
timeout: 30
# 重试间隔（秒）
retry_interval: 2
# 最大重试次数
max_retries: 3
# 任务配置
task:
# 默认任务类型：1-开发任务 2-测试任务 3-设计任务 4-部署任务 5-文档任务 6-维护任务 7-研究任务 8-会议沟通 9-售前支持 10-售后支持
type_id: 1
# 默认任务状态：1-新建 2-进行中 3-已解决 4-完成确认中 10-已完成
status_id: 1
# 默认紧急程度：1-每日观察 2-非常紧急 3-紧急处理 4-常规处理 5-迭代优化
level: 4
# 默认期望工时（小时）
excepthour: 14
# 默认责任人ID，陈泽键（441）
user_id: 441
# 默认创建人ID
createuser_id: 441
# 任务截止天数（当前时间+N天）
deadline_days: 7
# 需求关联配置
develop_search_str: "【新统一平台】公司内部安全测试扫描及修复验证"
# 缓存配置
cache_duration: 3600 # 基础数据缓存时长（秒），默认1小时
# 上传配置
upload:
# 是否上传截图
enabled: true
# 最多上传图片数量
max_images: 5
# 网盘配置（用于在任务描述中添加报告链接）
nas:
# 网盘基础路径
base_path: "\\\\192.168.9.9\\deploy\\18其它系统\\安全测试\\04新统一平台-安全报告"
AuxiliaryTool/ScriptTool/ApiSecurityTest/diagnose_erp.py 0 → 100644
浏览文件 @ 387ab3eb
# -*- coding: utf-8 -*-
"""
ERP对接诊断脚本
检查配置并测试ERP连接
"""
import sys
import yaml
from utils.logger import log
log.info("=" * 60)
log.info("ERP对接诊断工具")
log.info("=" * 60)
# 1. 检查配置文件
log.info("\n步骤 1/4: 检查配置文件...")
try:
with open('config.yaml', 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
erp_config = config.get('erp', {})
# 检查关键配置
log.info(f" - ERP启用状态: {erp_config.get('enabled', False)}")
log.info(f" - API基础URL: {erp_config.get('base_url', 'N/A')}")
log.info(f" - API密钥: {'*' * 10 if erp_config.get('api_key') and erp_config['api_key'] != 'your_api_key_here' else '未设置（需要配置）'}")
log.info(f" - 请求超时: {erp_config.get('timeout', 30)}秒")
log.info(f" - 重试次数: {erp_config.get('max_retries', 3)}")
task_config = erp_config.get('task', {})
log.info(f" - 责任人ID: {task_config.get('user_id', 'N/A')}")
log.info(f" - 创建人ID: {task_config.get('createuser_id', 'N/A')}")
if not erp_config.get('enabled', False):
log.warning(" ⚠ ERP对接未启用！请在config.yaml中设置 erp.enabled: true")
if not erp_config.get('api_key') or erp_config['api_key'] == 'your_api_key_here':
log.warning(" ⚠ API密钥未配置！请在config.yaml中设置 erp.api_key")
except Exception as e:
log.error(f" ✗ 配置文件读取失败: {e}")
sys.exit(1)
# 2. 测试ERP连接
log.info("\n步骤 2/4: 测试ERP连接...")
try:
from utils.erp_client import ERPClient
erp_client = ERPClient(config)
# 测试获取任务类型
log.info(" 正在测试API连接（获取任务类型）...")
task_types = erp_client.get_task_types()
if task_types:
log.info(f" ✓ 连接成功！获取到 {len(task_types)} 种任务类型")
for tt in task_types[:3]:
log.info(f" - {tt['id']}: {tt['name']}")
else:
log.error(" ✗ 连接失败：无法获取任务类型")
log.error(" 可能原因：")
log.error(" 1. API密钥无效")
log.error(" 2. 网络无法访问")
log.error(" 3. API端点地址错误")
except ImportError as e:
log.error(f" ✗ 模块导入失败: {e}")
except Exception as e:
log.error(f" ✗ 连接测试失败: {e}")
# 3. 测试获取人员列表
log.info("\n步骤 3/4: 测试获取人员列表...")
try:
staff_list = erp_client.get_staff_list()
if staff_list:
log.info(f" ✓ 获取到 {len(staff_list)} 人")
# 查找陈泽键
log.info(" 正在查找 '陈泽键'...")
chen_id = None
for staff in staff_list:
name = staff.get('name', '')
if '陈泽键' in name or name == '陈泽键':
chen_id = staff['id']
log.info(f" ✓ 找到陈泽键：ID = {chen_id}")
break
if chen_id is None:
log.warning(" ⚠ 未找到 '陈泽键'")
log.info(" 提示：可能是姓名不匹配，列出前10人供参考：")
for staff in staff_list[:10]:
log.info(f" ID: {staff['id']:3d} | 姓名: {staff['name']}")
else:
log.error(" ✗ 获取人员列表失败")
except Exception as e:
log.error(f" ✗ 获取人员列表失败: {e}")
# 4. 测试创建任务
log.info("\n步骤 4/4: 测试创建任务...")
try:
# 检查配置的ID是否有效
user_id = task_config.get('user_id')
createuser_id = task_config.get('createuser_id')
if not user_id or not createuser_id:
log.warning(" ⚠ 责任人ID或创建人ID未配置")
else:
# 验证ID是否存在
user_found = False
creator_found = False
for staff in staff_list:
if staff['id'] == user_id:
user_found = True
log.info(f" ✓ 责任人ID {user_id} 有效：{staff['name']}")
if staff['id'] == createuser_id:
creator_found = True
log.info(f" ✓ 创建人ID {createuser_id} 有效：{staff['name']}")
if not user_found:
log.warning(f" ⚠ 责任人ID {user_id} 在人员列表中不存在")
if not creator_found:
log.warning(f" ⚠ 创建人ID {createuser_id} 在人员列表中不存在")
except Exception as e:
log.error(f" ✗ 检查失败: {e}")
log.info("\n" + "=" * 60)
log.info("诊断完成")
log.info("=" * 60)
# 提供修复建议
log.info("\n修复建议：")
log.info("1. 如果API密钥未配置，请联系管理员获取有效的API密钥")
log.info("2. 如果未找到陈泽键，请确认姓名或手动设置ID")
log.info("3. 确保config.yaml中 erp.enabled 设置为 true")
AuxiliaryTool/ScriptTool/ApiSecurityTest/query_staff.py 0 → 100644
浏览文件 @ 387ab3eb
# -*- coding: utf-8 -*-
"""
查询ERP人员列表脚本
用于查找人员ID
"""
import sys
import yaml
from utils.erp_client import ERPClient
from utils.logger import log
# 加载配置
with open('config.yaml', 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
# 创建ERP客户端
erp_client = ERPClient(config)
# 获取人员列表
log.info("正在获取ERP人员列表...")
staff_list = erp_client.get_staff_list()
if staff_list:
log.info(f"✓ 共获取到 {len(staff_list)} 人")
# 打印所有人员
log.info("\n人员列表：")
log.info("-" * 60)
for staff in staff_list:
log.info(f"ID: {staff['id']:3d} | 姓名: {staff['name']:10s} | 部门ID: {staff.get('department_id', 'N/A')}")
log.info("-" * 60)
# 查找陈泽键
log.info("\n正在查找 '陈泽键'...")
found = False
for staff in staff_list:
if '陈泽键' in staff['name'] or staff['name'] == '陈泽键':
log.info(f"✓ 找到陈泽键：ID = {staff['id']}")
found = True
break
if not found:
log.warning("✗ 未找到 '陈泽键'")
log.info("提示：请检查人员姓名是否准确，可能需要使用部分匹配")
else:
log.error("✗ 获取人员列表失败")
AuxiliaryTool/ScriptTool/ApiSecurityTest/run_all.py
浏览文件 @ 387ab3eb
......@@ -6,6 +6,7 @@
import os
import sys
import shutil
import yaml
# 将项目根目录加入 Python 路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
......@@ -18,6 +19,16 @@ from utils.report_generator import ReportGenerator
# 网盘固定路径：安全测试报告上传目录
NAS_REPORT_DIR = r"\\192.168.9.9\deploy\18其它系统\安全测试\04新统一平台-安全报告"
# 加载配置文件
CONFIG = None
try:
with open('config.yaml', 'r', encoding='utf-8') as f:
CONFIG = yaml.safe_load(f)
log.info("配置文件加载成功")
except Exception as e:
log.warning(f"配置文件加载失败: {e}，将使用默认配置")
CONFIG = {}
def upload_to_nas(report_path):
"""
......@@ -54,13 +65,50 @@ def upload_to_nas(report_path):
def upload_to_erp(report_path):
"""
对接ERP创建任务跟踪（预留功能）
暂无对接，仅保留接口入口
对接ERP创建任务跟踪
参数:
report_path: 报告文件路径
返回:
bool: 是否成功创建任务
"""
log.info("ERP对接功能暂未实现，跳过")
try:
# 1. 检查是否启用ERP对接
if not CONFIG or not CONFIG.get('erp', {}).get('enabled', False):
log.info("ERP对接未启用，跳过任务创建")
return False
log.info("=" * 60)
log.info("ERP任务创建阶段")
log.info("=" * 60)
# 2. 导入ERP模块（延迟导入，避免影响主流程）
from utils.erp_client import ERPClient
from utils.task_creator import TaskCreator
# 3. 初始化ERP客户端
erp_client = ERPClient(CONFIG)
# 4. 创建任务创建器
task_creator = TaskCreator(erp_client, CONFIG)
# 5. 从报告创建任务
result = task_creator.create_task_from_report(report_path)
if result:
return True
else:
log.error("✗ ERP任务创建失败")
return False
except ImportError as e:
log.warning(f"ERP模块导入失败: {e}")
log.info("跳过ERP任务创建")
return False
except Exception as e:
log.error(f"✗ ERP任务创建异常: {e}")
return False
def run_all_tests():
......@@ -137,10 +185,11 @@ def run_all_tests():
log.info(f"总报告路径: {report_path}")
# 上传报告至网盘
log.info("阶段 4/4: 上传报告至网盘...")
log.info("阶段 4/5: 上传报告至网盘...")
upload_to_nas(report_path)
# ERP对接（预留）
# ERP对接
log.info("阶段 5/5: 创建ERP任务...")
upload_to_erp(report_path)
return report
......
AuxiliaryTool/ScriptTool/ApiSecurityTest/test_erp_creation.py 0 → 100644
浏览文件 @ 387ab3eb
# -*- coding: utf-8 -*-
"""
测试ERP任务创建功能
使用现有报告快速测试ERP对接
"""
import os
import sys
import yaml
from utils.logger import log
from utils.erp_client import ERPClient
from utils.task_creator import TaskCreator
log.info("=" * 60)
log.info("ERP任务创建测试")
log.info("=" * 60)
# 加载配置
with open('config.yaml', 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
# 检查ERP是否启用
if not config.get('erp', {}).get('enabled', False):
log.error("ERP对接未启用！请在config.yaml中设置 erp.enabled: true")
sys.exit(1)
# 查找最近的报告
reports_dir = "reports"
if not os.path.exists(reports_dir):
log.error(f"报告目录不存在: {reports_dir}")
log.info("请先运行安全测试生成报告：python run_all.py")
sys.exit(1)
# 获取最新的报告文件
report_files = [f for f in os.listdir(reports_dir) if f.endswith('.md')]
if not report_files:
log.error(f"没有找到报告文件: {reports_dir}/*.md")
log.info("请先运行安全测试生成报告：python run_all.py")
sys.exit(1)
# 按修改时间排序，获取最新的
report_files.sort(key=lambda x: os.path.getmtime(os.path.join(reports_dir, x)), reverse=True)
latest_report = os.path.join(reports_dir, report_files[0])
log.info(f"使用报告: {latest_report}")
log.info(f"报告时间: {report_files[0]}")
# 测试ERP任务创建
log.info("\n开始测试ERP任务创建...")
try:
# 初始化ERP客户端
erp_client = ERPClient(config)
# 创建任务创建器
task_creator = TaskCreator(erp_client, config)
# 从报告创建任务
result = task_creator.create_task_from_report(latest_report)
# 调试：打印 result 的类型和内容
log.info(f"[DEBUG] result type: {type(result)}")
log.info(f"[DEBUG] result keys: {result.keys() if isinstance(result, dict) else 'N/A'}")
if isinstance(result, dict):
log.info(f"[DEBUG] result.get('success'): {result.get('success')}")
if 'data' in result:
log.info(f"[DEBUG] result['data']: {result['data']}")
if result and isinstance(result, dict) and result.get('success') == 1:
# 从 data 字段中获取任务信息
data = result.get('data', {})
task_id = data.get('create')
task_name = data.get('name')
log.info("\n" + "=" * 60)
log.info("✅ 测试成功！ERP任务创建成功！")
log.info("=" * 60)
log.info(f" 任务ID: {task_id}")
log.info(f" 任务名称: {task_name}")
log.info(f" 查看链接: {config['erp']['base_url']}/task/detail/{task_id}")
log.info("=" * 60)
log.info("\n请登录ERP系统验证任务是否创建成功！")
else:
log.error("\n❌ 测试失败！ERP任务创建失败")
if result:
log.error(f" result内容: {result}")
else:
log.error(" result is None or empty")
except Exception as e:
log.error(f"\n❌ 测试异常: {e}")
import traceback
traceback.print_exc()
AuxiliaryTool/ScriptTool/ApiSecurityTest/utils/erp_client.py 0 → 100644
浏览文件 @ 387ab3eb
# -*- coding: utf-8 -*-
"""
ERP客户端模块
对接ERP任务管理接口，提供任务创建相关功能
"""
import requests
import time
from typing import Optional, Dict, List
from utils.logger import log
# API端点常量
TASK_TYPE_ENDPOINT = "/openclaw/tasktype"
TASK_STATUS_ENDPOINT = "/openclaw/taskstatus"
STAFF_ENDPOINT = "/openclaw/stuff"
DEVELOP_ENDPOINT = "/openclaw/develop"
TASK_ENDPOINT = "/openclaw/task"
UPLOAD_IMAGE_ENDPOINT = "/openclaw/upload/richtext"
class ERPClient:
"""ERP系统客户端"""
def __init__(self, config: Dict):
"""
初始化客户端
参数:
config: 配置字典（从config.yaml读取）
"""
erp_config = config.get('erp', {})
self.base_url = erp_config.get('base_url', '')
self.api_key = erp_config.get('api_key', '')
self.timeout = erp_config.get('timeout', 30)
self.retry_interval = erp_config.get('retry_interval', 2)
self.max_retries = erp_config.get('max_retries', 3)
# 缓存基础数据
self._cache = {}
self._cache_timestamps = {}
cache_duration = erp_config.get('task', {}).get('cache_duration', 3600)
self._cache_duration = cache_duration
def get_task_types(self) -> Optional[List[Dict]]:
"""
获取任务类型列表
返回:
任务类型列表，格式：[{"id": 1, "name": "开发任务"}, ...]
失败返回None
"""
result = self._get_with_cache('task_types', lambda: self._make_request('GET', TASK_TYPE_ENDPOINT))
if result and isinstance(result, dict) and 'data' in result:
return result.get('data')
return result
def get_task_statuses(self) -> Optional[List[Dict]]:
"""
获取任务状态列表
返回:
任务状态列表，格式：[{"id": 1, "name": "新建"}, ...]
失败返回None
"""
result = self._get_with_cache('task_statuses', lambda: self._make_request('GET', TASK_STATUS_ENDPOINT))
if result and isinstance(result, dict) and 'data' in result:
return result.get('data')
return result
def get_staff_list(self) -> Optional[List[Dict]]:
"""
获取人员列表
返回:
人员列表，格式：[{"id": 1, "name": "超管员", "department_id": null}, ...]
失败返回None
"""
result = self._get_with_cache('staff_list', lambda: self._make_request('GET', STAFF_ENDPOINT))
if result and isinstance(result, dict) and 'data' in result:
return result.get('data')
return result
def get_develop_list(self, page_num: int = 1, page_size: int = 20, search_str: str = "") -> Optional[Dict]:
"""
获取需求列表（支持搜索和分页）
参数:
page_num: 页码，默认1
page_size: 每页数量，默认20
search_str: 搜索字符串
返回:
包含data、totalRows等字段的字典
失败返回None
"""
params = {
'pageNum': page_num,
'pageSize': page_size
}
if search_str:
params['search_str'] = search_str
return self._make_request('GET', DEVELOP_ENDPOINT, params=params)
def create_task(self, task_data: Dict) -> Optional[Dict]:
"""
创建任务
参数:
task_data: 任务数据字典
返回:
创建结果，包含任务ID和名称
失败返回None
"""
result = self._make_request('POST', TASK_ENDPOINT, json=task_data)
return result
def upload_image(self, file_path: str) -> Optional[str]:
"""
上传图片
参数:
file_path: 图片文件路径
返回:
图片URL，失败返回None
"""
try:
with open(file_path, 'rb') as f:
files = {'file': f}
data = {'name': f'task_image_{int(time.time())}.png'}
result = self._make_request('POST', UPLOAD_IMAGE_ENDPOINT, files=files, data=data, is_file_upload=True)
if result and result.get('success') == 1:
return result.get('data', {}).get('url')
except Exception as e:
log.error(f"上传图片失败: {e}")
return None
def _get_with_cache(self, cache_key: str, fetch_func) -> Optional:
"""
带缓存的数据获取
参数:
cache_key: 缓存键名
fetch_func: 获取数据的函数
返回:
缓存的数据或新获取的数据
"""
# 检查缓存是否存在且未过期
if cache_key in self._cache:
timestamp = self._cache_timestamps.get(cache_key, 0)
if time.time() - timestamp < self._cache_duration:
log.debug(f"使用缓存数据: {cache_key}")
return self._cache[cache_key]
# 获取新数据
data = fetch_func()
if data is not None:
self._cache[cache_key] = data
self._cache_timestamps[cache_key] = time.time()
return data
def clear_cache(self):
"""清空所有缓存"""
self._cache.clear()
self._cache_timestamps.clear()
log.info("ERP基础数据缓存已清空")
def _make_request(self, method: str, endpoint: str, **kwargs) -> Optional[Dict]:
"""
统一的请求方法（含重试机制）
参数:
method: HTTP方法（GET/POST）
endpoint: API端点
**kwargs: 其他请求参数（params、json、files等）
返回:
响应数据字典，失败返回None
"""
url = f"{self.base_url}{endpoint}"
headers = {'X-Api-Key': self.api_key}
# 处理不同类型的请求参数
is_file_upload = kwargs.pop('is_file_upload', False)
if is_file_upload:
# 文件上传，不设置Content-Type
files = kwargs.get('files', {})
data = kwargs.get('data', {})
request_kwargs = {'files': files, 'data': data}
else:
# 普通JSON请求
headers['Content-Type'] = 'application/json'
json_data = kwargs.pop('json', None)
params = kwargs.pop('params', None)
request_kwargs = {}
if json_data:
request_kwargs['json'] = json_data
if params:
request_kwargs['params'] = params
# 重试机制
for attempt in range(self.max_retries):
try:
log.info(f"ERP API请求: {method} {url}")
# 打印请求详情（除了敏感信息）
if json_data:
log.info(f"请求参数（JSON）: {json_data}")
if params:
log.info(f"请求参数（URL）: {params}")
if headers:
# 隐藏API密钥的部分
safe_headers = headers.copy()
if 'X-Api-Key' in safe_headers:
key = safe_headers['X-Api-Key']
safe_headers['X-Api-Key'] = f"{key[:8]}...{key[-4:]}" if len(key) > 12 else "***"
log.info(f"请求头: {safe_headers}")
# 发送请求
if is_file_upload:
response = requests.post(
url,
headers=headers,
timeout=self.timeout,
**request_kwargs
)
elif method == 'GET':
response = requests.get(
url,
headers=headers,
timeout=self.timeout,
**request_kwargs
)
elif method == 'POST':
response = requests.post(
url,
headers=headers,
timeout=self.timeout,
**request_kwargs
)
else:
log.error(f"不支持的HTTP方法: {method}")
return None
# 检查响应状态码
if response.status_code == 200:
result = response.json()
# 详细日志：打印完整响应
log.info(f"响应状态码: {response.status_code}")
log.info(f"响应内容: {result}")
if result.get('success') == 1:
log.info(f"✓ ERP API请求成功: {endpoint}")
return result
else:
# 客户端错误（4xx）不重试
error_code = result.get('error', 0)
if 400 <= error_code < 500:
log.error(f"✗ ERP API请求失败（客户端错误 {error_code}）: {result.get('msg', '未知错误')}")
return None
# 服务端错误（5xx）重试
if attempt < self.max_retries - 1:
log.warning(f"ERP API服务端错误，{self.retry_interval}秒后重试({attempt+1}/{self.max_retries}): {result.get('msg')}")
time.sleep(self.retry_interval)
else:
log.error(f"✗ ERP API请求失败（已达最大重试次数）: {result.get('msg')}")
return None
else:
# HTTP错误
if attempt < self.max_retries - 1 and response.status_code >= 500:
log.warning(f"ERP API返回HTTP {response.status_code}，{self.retry_interval}秒后重试({attempt+1}/{self.max_retries})")
time.sleep(self.retry_interval)
else:
log.error(f"✗ ERP API HTTP错误: {response.status_code}")
return None
except requests.exceptions.Timeout:
if attempt < self.max_retries - 1:
log.warning(f"ERP API请求超时，{self.retry_interval}秒后重试({attempt+1}/{self.max_retries})")
time.sleep(self.retry_interval)
else:
log.error(f"✗ ERP API请求超时（已达最大重试次数）")
return None
except requests.exceptions.RequestException as e:
log.error(f"✗ ERP API请求异常: {e}")
return None
except Exception as e:
log.error(f"✗ ERP API未知异常: {e}")
return None
return None
AuxiliaryTool/ScriptTool/ApiSecurityTest/utils/logger.py
浏览文件 @ 387ab3eb
......@@ -4,7 +4,9 @@
"""
import os
import logging
import yaml
from datetime import datetime
from pathlib import Path
# 尝试导入 colorama，不存在则降级处理
try:
......@@ -15,6 +17,15 @@ except ImportError:
HAS_COLORAMA = False
def load_config():
"""加载配置文件"""
config_file = Path(__file__).parent.parent / 'config.yaml'
if config_file.exists():
with open(config_file, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
return {}
class ColoredFormatter(logging.Formatter):
"""带颜色的日志格式化器"""
......@@ -40,17 +51,28 @@ class ColoredFormatter(logging.Formatter):
return formatted
def setup_logger(name="ApiSecurityTest", log_dir=None):
def setup_logger(name="ApiSecurityTest", log_dir=None, config=None):
"""
创建并配置日志器
参数:
name: 日志器名称
log_dir: 日志文件输出目录，为None则不输出文件
log_dir: 日志文件输出目录，为None则从配置读取
config: 配置字典，为None则自动加载
返回:
logging.Logger: 配置好的日志器
"""
# 加载配置
if config is None:
config = load_config()
# 从配置获取日志目录
if log_dir is None:
log_config = config.get('log', {})
if log_config.get('file_enabled', False):
log_dir = log_config.get('output_dir', 'logs')
logger = logging.getLogger(name)
# 防止重复添加处理器
......@@ -80,6 +102,7 @@ def setup_logger(name="ApiSecurityTest", log_dir=None):
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter(fmt, date_fmt))
logger.addHandler(file_handler)
logger.info(f"日志文件: {log_file}")
return logger
......
AuxiliaryTool/ScriptTool/ApiSecurityTest/utils/report_parser.py 0 → 100644
浏览文件 @ 387ab3eb
# -*- coding: utf-8 -*-
"""
报告解析器模块
从Markdown格式的安全测试报告中提取关键信息
"""
import re
import os
from typing import Dict, Optional, List
from datetime import datetime
from utils.logger import log
class ReportParser:
"""安全测试报告解析器"""
def __init__(self, report_path: str):
"""
初始化解析器
参数:
report_path: 报告文件路径
"""
self.report_path = report_path
self.content = None
self.summary = {}
self.target_url = ""
self.start_time = None
self.end_time = None
def load(self) -> bool:
"""
加载报告文件
返回:
bool: 加载是否成功
"""
try:
if not os.path.isfile(self.report_path):
log.error(f"报告文件不存在: {self.report_path}")
return False
with open(self.report_path, 'r', encoding='utf-8') as f:
self.content = f.read()
log.info(f"成功加载报告文件: {self.report_path}")
return True
except Exception as e:
log.error(f"加载报告文件失败: {e}")
return False
def parse_summary(self) -> Dict:
"""
解析测试摘要
返回:
包含以下字段的字典：
- total: 总测试用例数
- high: 高危漏洞数
- medium: 中危漏洞数
- low: 低危漏洞数
- info: 信息类数
- safe: 安全项数
- target: 测试目标URL
- start_time: 测试开始时间
- end_time: 测试结束时间
"""
if not self.content:
log.error("报告内容未加载，请先调用load()方法")
return {}
summary = {
'total': 0,
'high': 0,
'medium': 0,
'low': 0,
'info': 0,
'safe': 0,
'target': '',
'start_time': '',
'end_time': ''
}
try:
# 提取测试目标（格式：> **测试目标**：https://192.168.5.44）
target_match = re.search(r'> \*\*测试目标\*\*[：:]\s*(https?://[^\s\n]+)', self.content)
if target_match:
summary['target'] = target_match.group(1).strip()
self.target_url = summary['target']
# 提取测试时间（格式：| 测试时间 | 2026-06-16 14:25:06 ~ 2026-06-16 14:27:38 |）
time_match = re.search(r'\|\s*测试时间\s*\|\s*(.+?~.+?)\s*\|', self.content)
if time_match:
time_range = time_match.group(1).strip()
# 分割开始和结束时间
times = re.split(r'\s*~\s*', time_range)
if len(times) == 2:
summary['start_time'] = times[0].strip()
summary['end_time'] = times[1].strip()
self.start_time = summary['start_time']
self.end_time = summary['end_time']
# 提取漏洞统计（格式：| 测试用例总数 | **145** |）
total_match = re.search(r'\|\s*测试用例总数\s*\|\s*\*\*(\d+)\*\*', self.content)
if total_match:
summary['total'] = int(total_match.group(1))
# 提取各等级漏洞数
high_match = re.search(r'\|\s*🔴 高危漏洞\s*\|\s*\*\*(\d+)\*\*', self.content)
if high_match:
summary['high'] = int(high_match.group(1))
medium_match = re.search(r'\|\s*🟠 中危漏洞\s*\|\s*\*\*(\d+)\*\*', self.content)
if medium_match:
summary['medium'] = int(medium_match.group(1))
low_match = re.search(r'\|\s*🟡 低危漏洞\s*\|\s*\*\*(\d+)\*\*', self.content)
if low_match:
summary['low'] = int(low_match.group(1))
info_match = re.search(r'\|\s*🔵 信息类\s*\|\s*\*\*(\d+)\*\*', self.content)
if info_match:
summary['info'] = int(info_match.group(1))
# 提取安全项（格式：| 🟢 已验证安全 | **123** |）
safe_match = re.search(r'\|\s*🟢 已验证安全\s*\|\s*\*\*(\d+)\*\*', self.content)
if safe_match:
summary['safe'] = int(safe_match.group(1))
# 如果还没提取到安全项，尝试另一种格式
if summary['safe'] == 0:
safe_match = re.search(r'\|\s*🟢 安全\s*\|\s*\*\*(\d+)\*\*', self.content)
if safe_match:
summary['safe'] = int(safe_match.group(1))
self.summary = summary
log.info(f"解析测试摘要: 总计={summary['total']}, 高危={summary['high']}, 中危={summary['medium']}, 低危={summary['low']}, 信息类={summary['info']}, 安全={summary['safe']}")
return summary
except Exception as e:
log.error(f"解析测试摘要失败: {e}")
return summary
def parse_vulnerabilities(self) -> List[Dict]:
"""
解析漏洞列表
返回:
漏洞列表，每个漏洞包含：
- id: 漏洞编号
- name: 漏洞名称
- level: 风险等级
- description: 漏洞描述
"""
vulnerabilities = []
if not self.content:
log.error("报告内容未加载，请先调用load()方法")
return vulnerabilities
try:
# 提取漏洞详情（假设格式为：| 漏洞编号 | 漏洞名称 | 风险等级 | 描述 |）
pattern = r'\|\s*([A-Z]{2}-\d+(?:\.\d+)?)\s*\|\s*([^|]+?)\s*\|\s*(🔴高危|🟠中危|🟡低危|🔵信息类)\s*\|\s*([^|]+?)\s*\|'
matches = re.findall(pattern, self.content)
for match in matches:
vulnerabilities.append({
'id': match[0].strip(),
'name': match[1].strip(),
'level': match[2].strip(),
'description': match[3].strip()
})
log.info(f"解析到 {len(vulnerabilities)} 个漏洞")
return vulnerabilities
except Exception as e:
log.error(f"解析漏洞列表失败: {e}")
return vulnerabilities
def get_high_risk_list(self) -> List[str]:
"""
获取高危漏洞列表
返回:
高危漏洞名称列表
"""
vulnerabilities = self.parse_vulnerabilities()
high_risk = [v['name'] for v in vulnerabilities if v['level'] == '🔴高危']
return high_risk
def get_target_ip(self) -> str:
"""
从目标URL中提取IP地址
返回:
IP地址字符串
"""
if self.target_url:
# 从URL中提取IP地址
ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', self.target_url)
if ip_match:
return ip_match.group(1)
return ""
def generate_task_name(self) -> str:
"""
生成ERP任务名称
格式：{目标IP}接口安全测试及漏洞修复_{日期}
示例：192.168.5.44接口安全测试及漏洞修复_20260624
返回:
任务名称字符串
"""
ip = self.get_target_ip()
date_str = datetime.now().strftime('%Y%m%d')
if ip:
return f"{ip}接口安全测试及漏洞修复_{date_str}"
else:
return f"接口安全测试及漏洞修复_{date_str}"
def generate_task_content(self, nas_report_path: str = "") -> str:
"""
生成ERP任务描述（HTML格式）
参数:
nas_report_path: 网盘报告路径（可选）
返回:
HTML格式的任务描述
"""
if not self.summary:
self.parse_summary()
html_parts = []
# 标题
html_parts.append("<h3>接口安全测试报告</h3>")
# 测试概览
html_parts.append("<h4>一、测试概览</h4>")
html_parts.append("<table border='1' cellpadding='5' cellspacing='0' style='border-collapse: collapse; width:100%;'>")
html_parts.append(f"<tr><td style='width:20%;'><strong>测试目标：</strong></td><td>{self.summary.get('target', 'N/A')}</td></tr>")
html_parts.append(f"<tr><td><strong>测试时间：</strong></td><td>{self.summary.get('start_time', 'N/A')} ~ {self.summary.get('end_time', 'N/A')}</td></tr>")
html_parts.append(f"<tr><td><strong>测试标准：</strong></td><td>OWASP API Security Top 10 (2019)</td></tr>")
html_parts.append(f"<tr><td><strong>测试工具：</strong></td><td>ApiSecurityTest (Python + requests + pycryptodome)</td></tr>")
html_parts.append("</table>")
# 漏洞统计
html_parts.append("<h4>二、漏洞统计</h4>")
html_parts.append("<table border='1' cellpadding='5' cellspacing='0' style='border-collapse: collapse;'>")
html_parts.append("<tr><th>总计</th><th>高危</th><th>中危</th><th>低危</th><th>信息类</th><th>安全</th></tr>")
html_parts.append(f"<tr><td>{self.summary.get('total', 0)}</td>")
html_parts.append(f"<td style='color:red; font-weight:bold;'>{self.summary.get('high', 0)}</td>")
html_parts.append(f"<td style='color:orange; font-weight:bold;'>{self.summary.get('medium', 0)}</td>")
html_parts.append(f"<td style='color:#FFD700; font-weight:bold;'>{self.summary.get('low', 0)}</td>")
html_parts.append(f"<td style='color:blue; font-weight:bold;'>{self.summary.get('info', 0)}</td>")
html_parts.append(f"<td style='color:green; font-weight:bold;'>{self.summary.get('safe', 0)}</td></tr>")
html_parts.append("</table>")
# 高危漏洞列表
high_risk_list = self.get_high_risk_list()
if high_risk_list:
html_parts.append("<h4>三、高危漏洞详情</h4>")
html_parts.append("<p><strong>本次测试发现以下高危漏洞，需优先处理：</strong></p>")
html_parts.append("<ol>")
for idx, vuln in enumerate(high_risk_list[:10], 1):
html_parts.append(f"<li style='color: red; margin-bottom: 5px;'><strong>{idx}. {vuln}</strong></li>")
if len(high_risk_list) > 10:
html_parts.append(f"<li>...及其他 {len(high_risk_list) - 10} 个高危漏洞</li>")
html_parts.append("</ol>")
# 中危漏洞统计
if self.summary.get('medium', 0) > 0:
html_parts.append("<h4>四、中危漏洞</h4>")
html_parts.append(f"<p>发现中危漏洞 <strong>{self.summary.get('medium', 0)}</strong> 个，建议优先修复。</p>")
# 测试建议
html_parts.append("<h4>五、修复建议</h4>")
html_parts.append("<ul>")
html_parts.append("<li><strong>优先处理高危漏洞：</strong>所有高危漏洞应在确认后24小时内修复</li>")
html_parts.append("<li><strong>中危漏洞：</strong>建议在本测试周期内完成修复</li>")
html_parts.append("<li><strong>低危漏洞：</strong>可在下一版本迭代中修复</li>")
html_parts.append("<li><strong>修复验证：</strong>漏洞修复完成后，请通知测试团队进行回归验证</li>")
html_parts.append("<li><strong>复查要求：</strong>所有漏洞修复后需要重新进行安全测试验证</li>")
html_parts.append("</ul>")
# 报告链接
if nas_report_path:
html_parts.append("<h4>六、测试报告</h4>")
html_parts.append(f"<p>详细测试报告已上传至网盘：</p>")
html_parts.append(f"<p><code>{nas_report_path}</code></p>")
html_parts.append("<p>请下载查看完整报告了解所有漏洞详情和修复建议。</p>")
# 紧急度说明
urgency_desc = {
5: "【最高优先级】有高危漏洞，需立即处理",
4: "【高优先级】有中危漏洞，建议优先处理",
3: "【中优先级】有低危漏洞，建议按计划修复",
2: "【正常优先级】仅有信息类问题",
1: "【低优先级】无安全漏洞"
}
calculated_level = self.calculate_urgency_level()
if calculated_level in urgency_desc:
html_parts.append(f"<p><strong>{urgency_desc[calculated_level]}</strong></p>")
# 联系信息
html_parts.append("<h4>七、联系方式</h4>")
html_parts.append("<p>如有疑问或需要协助，请联系测试团队。</p>")
return "\n".join(html_parts)
def calculate_urgency_level(self) -> int:
"""
根据漏洞数量计算紧急程度
返回:
紧急程度（1-5）：
5 - 有高危漏洞
4 - 有中危漏洞
3 - 有低危漏洞
2 - 仅有信息类
1 - 无漏洞
"""
if not self.summary:
self.parse_summary()
high = self.summary.get('high', 0)
medium = self.summary.get('medium', 0)
low = self.summary.get('low', 0)
info = self.summary.get('info', 0)
if high > 0:
return 5 # 有高危漏洞，最紧急
elif medium > 0:
return 4 # 有中危漏洞
elif low > 0:
return 3 # 有低危漏洞
elif info > 0:
return 2 # 仅有信息类
else:
return 1 # 无漏洞
AuxiliaryTool/ScriptTool/ApiSecurityTest/utils/task_creator.py 0 → 100644
浏览文件 @ 387ab3eb
# -*- coding: utf-8 -*-
"""
任务创建器模块
协调ERP客户端、报告解析器、缓存管理，完成任务创建
"""
import os
from datetime import datetime, timedelta
from typing import Optional, Dict
from utils.erp_client import ERPClient
from utils.report_parser import ReportParser
from utils.logger import log
class TaskCreator:
"""任务创建器"""
def __init__(self, erp_client: ERPClient, config: Dict):
"""
初始化任务创建器
参数:
erp_client: ERP客户端实例
config: 配置字典
"""
self.erp_client = erp_client
self.config = config
self.basic_data = {}
# 获取配置
erp_config = config.get('erp', {})
self.task_config = erp_config.get('task', {})
self.upload_config = erp_config.get('upload', {})
self.nas_config = erp_config.get('nas', {})
def create_task_from_report(self, report_path: str) -> Optional[Dict]:
"""
从测试报告创建ERP任务
参数:
report_path: 报告文件路径
返回:
创建的任务信息（包含任务ID），失败返回None
"""
log.info("=" * 60)
log.info("开始创建ERP任务")
log.info("=" * 60)
try:
# 1. 解析报告
log.info("步骤 1/5: 解析测试报告...")
parser = ReportParser(report_path)
if not parser.load():
log.error("加载报告文件失败")
return None
summary = parser.parse_summary()
if not summary:
log.error("解析报告摘要失败")
return None
log.info(f"✓ 测试目标: {summary.get('target', 'N/A')}")
log.info(f"✓ 测试时间: {summary.get('start_time', 'N/A')} ~ {summary.get('end_time', 'N/A')}")
log.info(f"✓ 漏洞统计: 高危={summary.get('high', 0)}, 中危={summary.get('medium', 0)}, 低危={summary.get('low', 0)}")
# 2. 获取ERP基础数据
log.info("步骤 2/5: 获取ERP基础数据...")
if not self._get_basic_data():
log.error("获取ERP基础数据失败")
return None
log.info(f"✓ 任务类型: {len(self.basic_data.get('task_types', []))}种")
log.info(f"✓ 任务状态: {len(self.basic_data.get('task_statuses', []))}种")
log.info(f"✓ 人员列表: {len(self.basic_data.get('staff_list', []))}人")
# 3. 搜索关联需求
log.info("步骤 3/5: 搜索关联需求...")
requirement = self._search_requirement()
if not requirement:
log.warning("未找到关联需求，将使用默认值")
else:
log.info(f"✓ 找到需求: {requirement.get('name', '')} (ID: {requirement.get('id', '')})")
# 4. 构建任务数据
log.info("步骤 4/5: 构建任务数据...")
task_data = self._build_task_data(parser, requirement)
# 5. 创建任务
log.info("步骤 5/5: 创建ERP任务...")
result = self.erp_client.create_task(task_data)
if result and result.get('success') == 1:
# 从 data 字段中获取任务信息
data = result.get('data', {})
task_id = data.get('create')
task_name = data.get('name')
log.info("=" * 60)
log.info("✓ ERP任务创建成功！")
log.info(f" - 任务ID: {task_id}")
log.info(f" - 任务名称: {task_name}")
log.info(f" - 任务链接: {self.erp_client.base_url}/task/detail/{task_id}")
log.info("=" * 60)
return result
else:
log.error("✗ ERP任务创建失败")
return None
except Exception as e:
log.error(f"✗ ERP任务创建异常: {e}")
return None
def _get_basic_data(self) -> bool:
"""
获取基础数据（使用缓存）
返回:
bool: 是否成功获取
"""
try:
# 获取任务类型
task_types = self.erp_client.get_task_types()
if not task_types:
log.warning("获取任务类型失败")
return False
# 获取任务状态
task_statuses = self.erp_client.get_task_statuses()
if not task_statuses:
log.warning("获取任务状态失败")
return False
# 获取人员列表
staff_list = self.erp_client.get_staff_list()
if not staff_list:
log.warning("获取人员列表失败")
return False
self.basic_data = {
'task_types': task_types,
'task_statuses': task_statuses,
'staff_list': staff_list
}
return True
except Exception as e:
log.error(f"获取基础数据异常: {e}")
return False
def _search_requirement(self) -> Optional[Dict]:
"""
搜索关联的需求
返回:
需求信息字典，未找到返回None
"""
try:
search_str = self.task_config.get('develop_search_str', '')
if not search_str:
return None
# 搜索需求
result = self.erp_client.get_develop_list(
page_num=1,
page_size=10,
search_str=search_str
)
# result 可能是列表或字典，统一处理
if result:
data_list = None
if isinstance(result, list):
# 如果直接返回列表
data_list = result
elif isinstance(result, dict) and 'data' in result:
# 如果返回字典，提取data字段
data_list = result['data']
# 从列表中获取第一个需求
if data_list and isinstance(data_list, list) and len(data_list) > 0:
return data_list[0]
return None
except Exception as e:
log.error(f"搜索需求异常: {e}")
return None
def _build_task_data(self, parser: ReportParser, requirement: Optional[Dict]) -> Dict:
"""
构建任务数据
参数:
parser: 报告解析器实例
requirement: 关联的需求信息
返回:
任务数据字典
"""
# 从配置获取默认值
task_name = parser.generate_task_name()
type_id = self.task_config.get('type_id', 2)
status_id = self.task_config.get('status_id', 1)
# 根据漏洞数量计算紧急程度
calculated_level = parser.calculate_urgency_level()
level = self.task_config.get('level', 3)
# 如果配置的level不是1-5，则使用计算值
if level < 1 or level > 5:
level = calculated_level
excepthour = self.task_config.get('excepthour', 8)
user_id = self.task_config.get('user_id', 20)
createuser_id = self.task_config.get('createuser_id', 5)
deadline_days = self.task_config.get('deadline_days', 3)
# 计算截止时间
deadline = datetime.now() + timedelta(days=deadline_days)
excepttime = deadline.strftime('%Y-%m-%d %H:%M:%S')
# 获取需求ID
develop_id = None
if requirement:
develop_id = requirement.get('id')
# 生成任务描述HTML
nas_path = self.nas_config.get('base_path', '')
report_filename = os.path.basename(parser.report_path)
nas_report_path = os.path.join(nas_path, report_filename) if nas_path else ""
content = parser.generate_task_content(nas_report_path)
# 构建任务数据
task_data = {
'name': task_name,
'type_id': type_id,
'status_id': status_id,
'level': level,
'excepthour': excepthour,
'excepttime': excepttime,
'content': content,
'user_id': user_id,
'createuser_id': createuser_id
}
# 添加可选字段
if develop_id:
task_data['develop_id'] = develop_id
log.info(f"任务数据:")
log.info(f" - 名称: {task_name}")
log.info(f" - 类型: {self._get_task_type_name(type_id)}")
log.info(f" - 状态: {self._get_task_status_name(status_id)}")
log.info(f" - 紧急程度: {level}")
log.info(f" - 期望工时: {excepthour}小时")
log.info(f" - 截止时间: {excepttime}")
log.info(f" - 责任人: {self._get_staff_name(user_id)}")
log.info(f" - 创建人: {self._get_staff_name(createuser_id)}")
if develop_id:
log.info(f" - 关联需求: {requirement.get('name', '')}")
return task_data
def _calculate_deadline(self) -> str:
"""
计算截止时间
返回:
格式化的截止时间字符串 (YYYY-MM-DD HH:mm:ss)
"""
deadline_days = self.task_config.get('deadline_days', 3)
deadline = datetime.now() + timedelta(days=deadline_days)
return deadline.strftime('%Y-%m-%d %H:%M:%S')
def _get_task_type_name(self, type_id: int) -> str:
"""根据ID获取任务类型名称"""
for task_type in self.basic_data.get('task_types', []):
if task_type.get('id') == type_id:
return task_type.get('name', '')
return f"未知类型({type_id})"
def _get_task_status_name(self, status_id: int) -> str:
"""根据ID获取任务状态名称"""
for status in self.basic_data.get('task_statuses', []):
if status.get('id') == status_id:
return status.get('name', '')
return f"未知状态({status_id})"
def _get_staff_name(self, staff_id: int) -> str:
"""根据ID获取人员姓名"""
for staff in self.basic_data.get('staff_list', []):
if staff.get('id') == staff_id:
return staff.get('name', '')
return f"未知人员({staff_id})"
Docs/PRD/接口安全测试/ERP对接/ERP任务模块_API接口文档.md 0 → 100644
浏览文件 @ 387ab3eb
# OpenClaw API - 任务接口文档
> 文档版本：v1.0
>
> 生成时间：2026-06-24
>
> 适用对象：第三方系统对接开发人员
---
## 目录
- [一、接口基础信息](#一接口基础信息)
- [二、接口清单](#二接口清单)
- [三、接口详细说明](#三接口详细说明)
- [3.1 获取任务类型](#31-获取任务类型)
- [3.2 获取任务状态](#32-获取任务状态)
- [3.3 获取人员列表](#33-获取人员列表)
- [3.4 获取需求列表](#34-获取需求列表)
- [3.5 上传图片](#35-上传图片)
- [3.6 创建任务](#36-创建任务)
- [3.7 查询任务列表](#37-查询任务列表)
- [3.8 更新任务](#38-更新任务)
- [3.9 删除任务](#39-删除任务)
- [四、重要说明](#四重要说明)
---
## 一、接口基础信息
### 1.1 环境地址
| 环境 | 地址 |
|------|------|
| 生产环境 | `https://office.ubainsyun.com:5082/api/uerp` |
| 测试环境 | `http://127.0.0.1:8002/uerp` |
### 1.2 认证方式
所有接口需要在请求头中携带 API Key：
```
X-Api-Key: your_api_key_here
```
### 1.3 请求格式
| 请求类型 | Content-Type |
|----------|--------------|
| JSON 请求 | `application/json` |
| 文件上传 | `multipart/form-data` |
### 1.4 响应格式
所有接口返回 JSON 格式：
```json
{
"success": 1, // 1=成功, 0=失败
"data": { ... }, // 成功时返回数据
"error": 40000014, // 失败时的错误码
"msg": "错误信息" // 失败时的错误描述
}
```
### 1.5 错误码说明
| 错误码 | 说明 | 解决方案 |
|--------|------|----------|
| 40000001 | 参数错误或业务异常 | 检查请求参数是否正确 |
| 40000014 | 无效的 API Key | 检查 X-Api-Key 是否正确 |
| 40000014 | API Key 已过期 | 联系管理员延长有效期 |
| 40000014 | 无权限访问该接口 | 联系管理员添加接口权限 |
---
## 二、接口清单
| 序号 | 接口名称 | 方法 | 路径 | 说明 |
|------|----------|------|------|------|
| 1 | 获取任务类型 | GET | `/openclaw/tasktype` | 获取任务分类列表 |
| 2 | 获取任务状态 | GET | `/openclaw/taskstatus` | 获取任务状态列表 |
| 3 | 获取人员列表 | GET | `/openclaw/stuff` | 获取可指派的人员 |
| 4 | 获取需求列表 | GET | `/openclaw/develop` | 获取需求列表（用于关联任务） |
| 5 | 上传图片 | POST | `/openclaw/upload/richtext` | 上传截图等图片 |
| 6 | 创建任务 | POST | `/openclaw/task` | 创建新任务 |
| 7 | 查询任务 | GET | `/openclaw/task` | 查询任务列表 |
| 8 | 更新任务 | PUT | `/openclaw/task` | 更新任务信息 |
| 9 | 删除任务 | DELETE | `/openclaw/task` | 删除任务（软删除） |
---
## 三、接口详细说明
### 3.1 获取任务类型
获取系统中可用的任务类型列表。
#### 请求信息
| 项目 | 值 |
|------|-----|
| 方法 | GET |
| 路径 | `/openclaw/tasktype` |
#### 请求头
```
X-Api-Key: your_api_key
```
#### 请求参数
无
#### 响应示例
```json
{
"success": 1,
"data": [
{"id": 1, "name": "开发任务"},
{"id": 2, "name": "测试任务"},
{"id": 3, "name": "设计任务"},
{"id": 4, "name": "部署任务"},
{"id": 5, "name": "文档任务"},
{"id": 6, "name": "维护任务"},
{"id": 7, "name": "研究任务"},
{"id": 8, "name": "会议沟通"},
{"id": 9, "name": "售前支持"},
{"id": 10, "name": "售后支持"}
]
}
```
#### 任务类型说明
| ID | 名称 | 说明 |
|----|------|------|
| 1 | 开发任务 | 开发相关任务 |
| 2 | 测试任务 | 测试相关任务 |
| 3 | 设计任务 | 设计相关任务 |
| 4 | 部署任务 | 部署相关任务 |
| 5 | 文档任务 | 文档编写任务 |
| 6 | 维护任务 | 系统维护任务 |
| 7 | 研究任务 | 技术研究任务 |
| 8 | 会议沟通 | 会议沟通任务 |
| 9 | 售前支持 | 售前支持任务 |
| 10 | 售后支持 | 售后支持任务 |
> **注意**：任务类型分为两组 — id ≤ 20 为标准任务类型，id > 20 为产品跟踪类型。
---
### 3.2 获取任务状态
获取任务状态流转列表。
#### 请求信息
| 项目 | 值 |
|------|-----|
| 方法 | GET |
| 路径 | `/openclaw/taskstatus` |
#### 请求头
```
X-Api-Key: your_api_key
```
#### 请求参数
无
#### 响应示例
```json
{
"success": 1,
"data": [
{"id": 1, "name": "新建"},
{"id": 2, "name": "进行中"},
{"id": 3, "name": "已解决"},
{"id": 4, "name": "完成确认中"},
{"id": 10, "name": "已完成"}
]
}
```
#### 状态说明
| 状态 ID | 状态名称 | 说明 |
|---------|----------|------|
| 1 | 新建 | 任务刚创建，等待开始 |
| 2 | 进行中 | 任务正在执行 |
| 3 | 已解决 | 任务已解决，待确认 |
| 4 | 完成确认中 | 等待验收确认 |
| 10 | 已完成 | 任务已关闭完成 |
#### 状态流转图
```
┌────────┐ ┌──────────┐ ┌────────┐ ┌──────────────┐ ┌────────┐
│ 新建 │ ──→ │ 进行中 │ ──→ │ 已解决 │ ──→ │ 完成确认中 │ ──→ │ 已完成 │
│ (1) │ │ (2) │ │ (3) │ │ (4) │ │ (10) │
└────────┘ └──────────┘ └────────┘ └──────────────┘ └────────┘
```
> **注意**：
> - 任务状态分为两组 — id ≤ 20 为标准状态，id > 20 为产品跟踪状态
> - `status_id < 10` 表示未完成任务
> - `status_id ≥ 10` 表示已完成任务
---
### 3.3 获取人员列表
获取系统中所有人员，用于指派任务责任人。
#### 请求信息
| 项目 | 值 |
|------|-----|
| 方法 | GET |
| 路径 | `/openclaw/stuff` |
#### 请求头
```
X-Api-Key: your_api_key
```
#### 请求参数
无
#### 响应示例
```json
{
"success": 1,
"data": [
{"id": 1, "name": "超管员", "department_id": null},
{"id": 20, "name": "郑晓兵", "department_id": 94020884},
{"id": 24, "name": "黄史恭", "department_id": 596543063},
{"id": 31, "name": "欧阳友平", "department_id": 363531662}
]
}
```
#### 响应字段说明
| 字段 | 类型 | 说明 |
|------|------|------|
| id | int | 人员 ID |
| name | string | 姓名 |
| department_id | int | 部门 ID |
---
### 3.4 获取需求列表
获取系统中所有需求，用于创建任务时关联需求（`develop_id` 必填）。
#### 请求信息
| 项目 | 值 |
|------|-----|
| 方法 | GET |
| 路径 | `/openclaw/develop` |
#### 请求头
```
X-Api-Key: your_api_key
```
#### 请求参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| pageNum | int | 否 | 页码，默认 1 |
| pageSize | int | 否 | 每页数量，默认 10 |
| id | int | 否 | 需求 ID（精确查询） |
| search_str | string | 否 | 全局搜索（需求名/描述/项目名/创建人/责任人） |
| project_id | int | 否 | 按项目 ID 过滤 |
| product_id | int | 否 | 按产品 ID 过滤 |
| module_id | int | 否 | 按模块 ID 过滤 |
| status_id | int | 否 | 按状态 ID 过滤 |
| dutyuser_id | int | 否 | 按责任人 ID 过滤 |
| requireuser_id | int | 否 | 按需求人 ID 过滤 |
| search_my | bool | 否 | 仅查询"我负责的"需求 |
| search_unfinish | bool | 否 | 包含已完成需求（默认不含 status_id ≥ 10） |
| group_value | string | 否 | 按分组标识过滤 |
#### 请求示例
```bash
curl -X GET "http://127.0.0.1:8002/uerp/openclaw/develop?pageNum=1&pageSize=10" \
-H "X-Api-Key: your_api_key"
```
#### 响应示例
```json
{
"success": 1,
"pageSize": 10,
"pageNum": 1,
"totalPages": 19,
"totalRows": 183,
"data": [
{
"id": 215,
"name": "测试需求",
"requirement": "<p>测试需求描述</p>",
"product_id": 1,
"product_name": "预约系统",
"module_id": 1,
"module_name": "标准版2.0红线",
"status_id": 1,
"status_name": "方案输出",
"level_id": 5,
"level_name": "每日观察",
"project_id": 1702,
"project_name": "演示测试项目2",
"createuser_id": 20,
"createuser_name": "郑晓兵",
"requireuser_id": 20,
"requireuser_name": "郑晓兵",
"dutyuser_id": 20,
"dutyuser_name": "郑晓兵",
"createdate": "2024-11-22",
"updatedate": "2024-11-22",
"exceptdate": "2024-11-22",
"closedate": "",
"task_count": 1,
"task_unfinish": 1,
"group_value": "cyzx"
}
]
}
```
#### 响应字段说明
| 字段 | 类型 | 说明 |
|------|------|------|
| id | int | 需求 ID（创建任务时的 `develop_id`） |
| name | string | 需求名称 |
| requirement | string | 需求描述（HTML 格式） |
| product_id | int | 产品 ID |
| product_name | string | 产品名称 |
| module_id | int | 模块 ID |
| module_name | string | 模块名称 |
| status_id | int | 状态 ID |
| status_name | string | 状态名称 |
| level_id | int | 等级 ID |
| level_name | string | 等级名称 |
| project_id | int | 关联项目 ID |
| project_name | string | 关联项目名称 |
| createuser_id | int | 创建人 ID |
| createuser_name | string | 创建人姓名 |
| requireuser_id | int | 需求人 ID |
| requireuser_name | string | 需求人姓名 |
| dutyuser_id | int | 责任人 ID |
| dutyuser_name | string | 责任人姓名 |
| createdate | string | 创建时间 |
| updatedate | string | 更新时间 |
| exceptdate | string | 截止时间 |
| closedate | string | 关闭时间 |
| task_count | int | 关联任务总数 |
| task_unfinish | int | 未完成任务数 |
| group_value | string | 分组标识 |
---
### 3.5 上传图片
上传任务截图等图片，返回完整的图片访问 URL。
#### 请求信息
| 项目 | 值 |
|------|-----|
| 方法 | POST |
| 路径 | `/openclaw/upload/richtext` |
| Content-Type | `multipart/form-data` |
#### 请求头
```
X-Api-Key: your_api_key
```
#### 请求参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| file | file | 是 | 图片文件 |
| name | string | 否 | 文件名称 |
#### 请求示例 (cURL)
```bash
curl -X POST "https://office.ubainsyun.com:5015/uerp/openclaw/upload/richtext" \
-H "X-Api-Key: your_api_key" \
-F "file=@/path/to/screenshot.png" \
-F "name=task_screenshot.png"
```
#### 响应示例
```json
{
"success": 1,
"data": {
"id": 12444,
"name": "task_screenshot.png",
"path": "upload/20260317223125747.png",
"url": "https://office.ubainsyun.com:5015/upload/20260317223125747.png",
"type": ".png",
"size": 209
}
}
```
#### 响应字段说明
| 字段 | 类型 | 说明 |
|------|------|------|
| id | int | 文件 ID |
| name | string | 文件名称 |
| path | string | 服务器存储路径 |
| url | string | **完整访问 URL，可直接用于富文本** |
| type | string | 文件扩展名 |
| size | int | 文件大小（字节） |
> **注意**：此接口与 BUG 接口共用。
---
### 3.6 创建任务 ⭐
创建一条新的任务记录。
#### 请求信息
| 项目 | 值 |
|------|-----|
| 方法 | POST |
| 路径 | `/openclaw/task` |
| Content-Type | `application/json` |
#### 请求头
```
X-Api-Key: your_api_key
Content-Type: application/json
```
#### 请求参数（必填）
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| name | string | 是 | 任务名称（最长128字符） |
| type_id | int | 是 | 任务类型 ID |
| status_id | int | 是 | 任务状态 ID |
| level | int | 是 | 紧急程度（1-5） |
| excepthour | int | 是 | 期望工时（小时） |
| excepttime | string | 是 | 截止时间，格式 `YYYY-MM-DD HH:mm:ss` |
| content | string | 是 | 任务描述（HTML 格式） |
| user_id | int | 是 | 主责任人 ID |
| createuser_id | int | 是 | 下单人员 ID（创建人） |
| develop_id | int | 是 | 关联的需求 ID |
#### 请求参数（可选）
| 参数 | 类型 | 说明 |
|------|------|------|
| planhour | int | 预计工时（小时），status_id > 1 时必填 |
| starttime | string | 开始时间，`YYYY-MM-DD HH:mm:ss`，status_id > 1 时必填 |
| endtime | string | 结束时间，`YYYY-MM-DD HH:mm:ss`，status_id > 1 时必填 |
| actualhour | int | 实际工时（小时），默认 0 |
| project_id | int | 关联的项目 ID |
| descript | string | 备注说明 |
| process | int | 当前进度（0-100），默认 0 |
| sort | int | 排序值，默认 0 |
| copyuserList | array | 抄送人员 ID 列表，如 `[{"id": 20}, {"id": 31}]` |
#### 紧急程度 (level) 说明
| level | 名称 | 说明 |
|-------|------|------|
| 1 | 每日观察 | 每日跟踪进度（最低优先级） |
| 2 | 非常紧急 | 需要立即处理 |
| 3 | 紧急处理 | 优先处理 |
| 4 | 常规处理 | 正常排期处理 |
| 5 | 迭代优化 | 后续版本处理（最高优先级） |
#### 请求示例
```json
{
"name": "用户登录模块接口开发",
"type_id": 1,
"status_id": 1,
"level": 3,
"excepthour": 8,
"excepttime": "2026-06-25 18:00:00",
"content": "<h4>任务描述</h4><p>完成用户登录模块的后端接口开发</p><h4>技术要求</h4><ul><li>使用 RESTful API 规范</li><li>JWT token 认证</li><li>接口文档同步更新</li></ul><h4>参考截图</h4><p><img src=\"https://office.ubainsyun.com:5015/upload/20260617223125747.png\" alt=\"接口设计图\"/></p>",
"user_id": 20,
"createuser_id": 5,
"develop_id": 128,
"planhour": 6,
"project_id": 42,
"descript": "优先级较高的开发任务"
}
```
#### 响应示例
```json
{
"success": 1,
"data": {
"create": 1024,
"name": "用户登录模块接口开发"
}
}
```
#### 响应字段说明
| 字段 | 类型 | 说明 |
|------|------|------|
| create | int | 新创建的任务 ID |
| name | string | 任务名称 |
---
### 3.7 查询任务列表
查询任务列表，支持分页和过滤。
#### 请求信息
| 项目 | 值 |
|------|-----|
| 方法 | GET |
| 路径 | `/openclaw/task` |
#### 请求头
```
X-Api-Key: your_api_key
```
#### 请求参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| pageNum | int | 否 | 页码，默认 1 |
| pageSize | int | 否 | 每页数量，默认 10 |
| id | int | 否 | 任务 ID |
| name | string | 否 | 任务名称（模糊搜索） |
| search_str | string | 否 | 全局搜索（匹配任务名/需求名/项目名/创建人/责任人） |
| status_id | int | 否 | 状态 ID |
| user_id | int | 否 | 责任人 ID |
| createuser_id | int | 否 | 创建人 ID |
| develop_id | int | 否 | 关联需求 ID |
| search_my | bool | 否 | 仅查询"我负责的"任务 |
| search_create | bool | 否 | 仅查询"我创建的"任务 |
| search_delay | bool | 否 | 仅查询"已延期"任务 |
| search_unfinish | bool | 否 | 包含已完成任务（默认不含 status_id ≥ 10） |
| search_product | int | 否 | 按产品 ID 过滤 |
| search_module | int | 否 | 按模块 ID 过滤 |
#### 请求示例
```bash
curl -X GET "https://office.ubainsyun.com:5015/uerp/openclaw/task?pageNum=1&pageSize=10&search_my=true" \
-H "X-Api-Key: your_api_key"
```
#### 响应示例
```json
{
"success": 1,
"pageSize": 10,
"pageNum": 1,
"totalPages": 15,
"totalRows": 148,
"data": [
{
"id": 1024,
"name": "用户登录模块接口开发",
"type_id": 1,
"type_name": "开发任务",
"status_id": 1,
"status_name": "新建",
"level": 3,
"content": "<h4>任务描述</h4><p>完成用户登录模块的后端接口开发</p>...",
"excepthour": 8,
"planhour": 6,
"actualhour": 0,
"remainhour": 0,
"process": 0,
"sort": 0,
"starttime": null,
"endtime": null,
"excepttime": "2026-06-25 18:00:00",
"user_id": 20,
"user_name": "郑晓兵",
"createuser_id": 5,
"createuser_name": "张三",
"updatetime": "2026-06-17 14:30:00",
"develop_id": 128,
"develop_name": "用户系统优化需求",
"project_id": 42,
"project_name": "ERP管理系统",
"product_name": "ERP系统",
"module_name": "用户管理",
"level_name": "3级",
"develop_requirement": "优化用户登录流程",
"develop_dutyuser_name": "李四",
"develop_requireuser_name": "王五",
"total_list": [],
"taskpoint_list": [],
"copyuserList": [],
"copyuser_name": ""
}
],
"projectList": [
{"value": 42, "text": "ERP管理系统"}
],
"createUserList": [
{"value": 5, "text": "张三"}
],
"userList": [
{"value": 20, "text": "郑晓兵"}
],
"totalData": {
"unfinish_count": 120,
"delay_count": 15,
"point_count": 3
}
}
```
#### 响应字段说明
| 字段 | 类型 | 说明 |
|------|------|------|
| id | int | 任务 ID |
| name | string | 任务名称 |
| type_id | int | 类型 ID |
| type_name | string | 类型名称 |
| status_id | int | 状态 ID |
| status_name | string | 状态名称 |
| level | int | 紧急程度（1-5） |
| content | string | 任务描述（HTML） |
| excepthour | int | 期望工时（小时） |
| planhour | int | 预计工时（小时） |
| actualhour | int | 实际工时（小时） |
| remainhour | int | 剩余工时（小时） |
| process | int | 当前进度（0-100） |
| starttime | string | 开始时间 |
| endtime | string | 结束时间 |
| excepttime | string | 截止时间 |
| user_id | int | 责任人 ID |
| user_name | string | 责任人姓名 |
| createuser_id | int | 创建人 ID |
| createuser_name | string | 创建人姓名 |
| updatetime | string | 更新时间 |
| develop_id | int | 关联需求 ID |
| develop_name | string | 关联需求名称 |
| project_id | int | 关联项目 ID |
| project_name | string | 关联项目名称 |
| totalData.unfinish_count | int | 未完成任务数量 |
| totalData.delay_count | int | 延期任务数量 |
---
### 3.8 更新任务
更新已有任务的信息。
#### 请求信息
| 项目 | 值 |
|------|-----|
| 方法 | PUT |
| 路径 | `/openclaw/task` |
| Content-Type | `application/json` |
#### 请求头
```
X-Api-Key: your_api_key
Content-Type: application/json
```
#### 请求参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| id | int | 是 | 任务 ID |
| name | string | 否 | 任务名称 |
| type_id | int | 否 | 任务类型 ID |
| status_id | int | 否 | 任务状态 ID |
| level | int | 否 | 紧急程度（1-5） |
| content | string | 否 | 任务描述 |
| excepthour | int | 否 | 期望工时 |
| planhour | int | 否 | 预计工时 |
| actualhour | float | 否 | 实际工时 |
| remainhour | int | 否 | 剩余工时 |
| starttime | string | 否 | 开始时间 |
| endtime | string | 否 | 结束时间 |
| excepttime | string | 否 | 截止时间 |
| process | int | 否 | 当前进度（0-100） |
| user_id | int | 否 | 责任人 ID |
| createuser_id | int | 否 | 创建人 ID |
| develop_id | int | 否 | 关联需求 ID |
| project_id | int | 否 | 关联项目 ID |
| descript | string | 否 | 本次更新说明（记录到历史） |
| copyuserList | array | 否 | 抄送人员列表 |
#### 请求示例 - 开始任务
```json
{
"id": 1024,
"status_id": 2,
"starttime": "2026-06-17 14:00:00",
"endtime": "2026-06-25 18:00:00",
"descript": "已开始开发工作"
}
```
#### 请求示例 - 完成任务
```json
{
"id": 1024,
"status_id": 4,
"process": 100,
"actualhour": 10,
"descript": "登录接口开发完成，已自测通过"
}
```
#### 请求示例 - 关闭任务
```json
{
"id": 1024,
"status_id": 10,
"descript": "验收通过，任务关闭"
}
```
#### 响应示例
```json
{
"success": 1,
"data": {
"update": 1024
}
}
```
---
### 3.9 删除任务
删除任务（软删除，设置 delflag=1）。
#### 请求信息
| 项目 | 值 |
|------|-----|
| 方法 | DELETE |
| 路径 | `/openclaw/task` |
| Content-Type | `application/json` |
#### 请求头
```
X-Api-Key: your_api_key
Content-Type: application/json
```
#### 请求参数
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
| id | int | 是 | 任务 ID |
#### 请求示例
```json
{
"id": 1024
}
```
#### 响应示例
```json
{
"success": 1,
"data": {
"delete": 1024
}
}
```
---
## 四、重要说明
### 4.1 必填字段规则
**创建任务（status_id=1）必填字段：**
- `name` - 任务名称
- `type_id` - 任务类型 ID
- `status_id` - 任务状态 ID
- `level` - 紧急程度
- `excepthour` - 期望工时
- `excepttime` - 截止时间
- `content` - 任务描述
- `user_id` - 主责任人 ID
- `createuser_id` - 下单人员 ID
- `develop_id` - 关联需求 ID
**状态 > 1 时额外必填：**
- `planhour` - 预计工时
- `starttime` - 开始时间
- `endtime` - 结束时间
**状态 ≥ 4 时必填：**
- `process` 必须为 100
### 4.2 人员角色区分
| 字段 | 说明 |
|------|------|
| user_id | 任务的责任人（处理人） |
| createuser_id | 下单人员（创建人） |
两者是不同的角色，需正确区分。
### 4.3 工时字段含义
| 字段 | 说明 |
|------|------|
| excepthour | 期望工时（计划需要的时间） |
| planhour | 预计工时（实际排期的时间） |
| actualhour | 实际工时（实际花费的时间） |
### 4.4 软删除说明
删除操作为软删除（delflag=1），不会从数据库中物理删除记录。
### 4.5 关联需求
`develop_id` 是必填字段，任务必须关联到需求（Develop）。可通过 `GET /openclaw/develop` 接口获取需求列表。
### 4.6 富文本内容
`content` 字段支持 HTML 格式，建议包含：
- 任务描述
- 技术要求（列表）
- 截图（img 标签）
- 参考资料（链接）
### 4.7 完整调用流程
```
┌─────────────────────────────────────────────────────────────────┐
│ 创建任务完整流程 │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 1. 获取基础数据（并行获取） │
│ - GET /openclaw/tasktype (任务类型) │
│ - GET /openclaw/taskstatus (任务状态) │
│ - GET /openclaw/stuff (人员列表) │
│ - GET /openclaw/develop (需求列表) │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 2. 上传截图（可选） │
│ POST /openclaw/upload/richtext │
│ 返回图片 URL 用于富文本内容 │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 3. 创建任务 │
│ POST /openclaw/task │
│ 提交：名称、类型、状态、紧急程度、工时、截止时间、 │
│ 责任人、下单人、关联需求、任务描述 │
└─────────────────────────────────────────────────────────────────┘
```
### 4.8 最小调用示例
如果已知类型、状态、紧急程度 ID，只需一步创建任务：
```bash
curl -X POST "https://office.ubainsyun.com:5015/uerp/openclaw/task" \
-H "X-Api-Key: your_api_key" \
-H "Content-Type: application/json" \
-d '{
"name": "示例任务",
"type_id": 1,
"status_id": 1,
"level": 4,
"excepthour": 4,
"excepttime": "2026-06-30 18:00:00",
"content": "<p>任务描述内容</p>",
"user_id": 20,
"createuser_id": 5,
"develop_id": 128
}'
```
---
## 附录
### A. 与 BUG 接口的关系
任务接口与 BUG 接口共用的资源：
| 共用接口 | 路径 | 说明 |
|----------|------|------|
| 人员列表 | `/openclaw/stuff` | 获取可指派的人员 |
| 图片上传 | `/openclaw/upload/richtext` | 上传截图等图片 |
### B. API Key 权限
确保 API Key 已添加以下权限：
- `task` - 任务管理
- `tasktype` - 任务类型
- `taskstatus` - 任务状态
---
**如有问题，请联系 ERP 系统管理员获取技术支持。**
Docs/PRD/接口安全测试/ERP对接/_PRD_ERP对接创建任务_需求文档.md 0 → 100644
浏览文件 @ 387ab3eb
# ERP创建任务对接需求文档
## 代码路径
- 代码路径：[AuxiliaryTool/ERPTaskCreation]
## ERP对接文档
- 需求文档路径：[Docs/PRD/ERP对接/PRD_测试列表_任务接口_例子说明.md]
## 功能需求
### 功能目标
**目标：** 对接ERP任务管理接口，实现通过程序创建开发任务，支持设置任务基本信息、指派人员、关联需求、上传截图等功能。
### 需求描述
#### 1. 获取基础数据
在创建任务前，需要获取以下基础数据供用户选择：
- **获取任务类型**
- 接口：`GET /openclaw/tasktype`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/tasktype`
- X-Api-Key: 保持原有的key传入
- 返回数据示例：
```json
{
"success": 1,
"data": [
{"id": 1, "name": "开发任务"},
{"id": 2, "name": "测试任务"},
{"id": 3, "name": "设计任务"},
{"id": 4, "name": "部署任务"},
{"id": 5, "name": "文档任务"},
{"id": 6, "name": "维护任务"},
{"id": 7, "name": "研究任务"},
{"id": 8, "name": "会议沟通"},
{"id": 9, "name": "售前支持"},
{"id": 10, "name": "售后支持"}
]
}
```
- 用途：用户选择任务类型
- **获取任务状态**
- 接口：`GET /openclaw/taskstatus`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/taskstatus`
- X-Api-Key: 保持原有的key传入
- 返回数据示例：
```json
{
"success": 1,
"data": [
{"id": 1, "name": "新建"},
{"id": 2, "name": "进行中"},
{"id": 3, "name": "已解决"},
{"id": 4, "name": "完成确认中"},
{"id": 10, "name": "已完成"}
]
}
```
- 用途：用户选择任务初始状态（通常选择"新建"）
- **获取人员列表**
- 接口：`GET /openclaw/stuff`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/stuff`
- X-Api-Key: 保持原有的key传入
- 返回数据示例：
```json
{
"success": 1,
"data": [
{"id": 1, "name": "超管员", "department_id": null},
{"id": 20, "name": "郑晓兵", "department_id": 94020884},
{"id": 24, "name": "黄史恭", "department_id": 596543063},
{"id": 31, "name": "欧阳友平", "department_id": 363531662}
]
}
```
- 用途：用户选择任务责任人（user_id）和下单人员（createuser_id）
- **获取需求列表**
- 接口：`GET /openclaw/develop`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/develop`
- X-Api-Key: 保持原有的key传入
- 查询参数：
- `pageNum`: 页码，默认 1
- `pageSize`: 每页数量，默认 10
- `search_str`: 全局搜索（需求名/描述/项目名/创建人/责任人）
- 返回数据示例：
```json
{
"success": 1,
"pageSize": 10,
"pageNum": 1,
"totalRows": 183,
"data": [
{
"id": 215,
"name": "测试需求",
"project_id": 1702,
"project_name": "演示测试项目2",
"dutyuser_id": 20,
"dutyuser_name": "郑晓兵"
}
]
}
```
- 用途：用户选择要关联的需求（develop_id 为必填）
#### 2. 上传截图（可选）
- 上传任务截图接口：
- 接口：`POST /openclaw/upload/richtext`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/upload/richtext`
- X-Api-Key: 保持原有的key传入
- Content-Type: `multipart/form-data`
- 请求示例：
```bash
curl -X POST "https://office.ubainsyun.com:5082/api/uerp/openclaw/upload/richtext" \
-H "X-Api-Key: your_api_key" \
-F "file=@/path/to/screenshot.png" \
-F "name=task_screenshot.png"
```
- 返回数据示例：
```json
{
"success": 1,
"data": {
"id": 12444,
"name": "task_screenshot.png",
"url": "https://office.ubainsyun.com:5015/upload/20260317223125747.png"
}
}
```
- 用途：上传截图后获取 URL，用于任务描述中的富文本内容
#### 3. 创建任务（核心功能）
- 创建任务接口：
- 接口：`POST /openclaw/task`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/task`
- X-Api-Key: 保持原有的key传入
- Content-Type: `application/json`
**必填参数（通过用户输入或选择）：**
| 参数 | 类型 | 说明 | 数据来源 |
|------|------|------|----------|
| name | string | 任务名称（最长128字符） | 用户输入 |
| type_id | int | 任务类型 ID | 从任务类型列表中选择 |
| status_id | int | 任务状态 ID | 从任务状态列表中选择（默认1-新建） |
| level | int | 紧急程度（1-5） | 用户选择或输入 |
| excepthour | int | 期望工时（小时） | 用户输入 |
| excepttime | string | 截止时间 `YYYY-MM-DD HH:mm:ss` | 用户选择或输入 |
| content | string | 任务描述（HTML 格式） | 用户输入，支持富文本 |
| user_id | int | 主责任人 ID | 从人员列表中选择 |
| createuser_id | int | 下单人员 ID（创建人） | 从人员列表中选择 |
| develop_id | int | 关联的需求 ID | 从需求列表中选择 |
**可选参数：**
| 参数 | 类型 | 说明 | 数据来源 |
|------|------|------|----------|
| planhour | int | 预计工时（小时） | 用户输入，status_id > 1 时必填 |
| starttime | string | 开始时间 `YYYY-MM-DD HH:mm:ss` | 用户选择，status_id > 1 时必填 |
| endtime | string | 结束时间 `YYYY-MM-DD HH:mm:ss` | 用户选择，status_id > 1 时必填 |
| actualhour | int | 实际工时（小时） | 用户输入，默认 0 |
| project_id | int | 关联的项目 ID | 用户选择 |
| descript | string | 备注说明 | 用户输入 |
| process | int | 当前进度（0-100） | 用户输入，默认 0 |
| sort | int | 排序值 | 用户输入，默认 0 |
| copyuserList | array | 抄送人员 ID 列表 | 从人员列表中多选 |
**紧急程度（level）对照表：**
| level | 名称 | 说明 |
|-------|------|------|
| 1 | 每日观察 | 每日跟踪进度（最低优先级） |
| 2 | 非常紧急 | 需要立即处理 |
| 3 | 紧急处理 | 优先处理 |
| 4 | 常规处理 | 正常排期处理 |
| 5 | 迭代优化 | 后续版本处理（最高优先级） |
**请求示例：**
```bash
curl -X POST "https://office.ubainsyun.com:5082/api/uerp/openclaw/task" \
-H "X-Api-Key: your_api_key" \
-H "Content-Type: application/json" \
-d '{
"name": "用户登录模块接口开发",
"type_id": 1,
"status_id": 1,
"level": 3,
"excepthour": 8,
"excepttime": "2026-06-25 18:00:00",
"content": "<h4>任务描述</h4><p>完成用户登录模块的后端接口开发</p><h4>技术要求</h4><ul><li>使用 RESTful API 规范</li><li>JWT token 认证</li><li>接口文档同步更新</li></ul><h4>参考截图</h4><p><img src=\"https://office.ubainsyun.com:5015/upload/20260617223125747.png\" alt=\"接口设计图\"/></p>",
"user_id": 20,
"createuser_id": 5,
"develop_id": 128,
"planhour": 6,
"project_id": 42,
"descript": "优先级较高的开发任务"
}'
```
**返回数据示例：**
```json
{
"success": 1,
"data": {
"create": 1024,
"name": "用户登录模块接口开发"
}
}
```
### 实现要求
1. **基础数据缓存**
- 获取任务类型、任务状态、人员列表后应进行缓存
- 避免每次创建任务都重复请求基础数据
- 建议缓存时长：1小时
2. **需求列表搜索**
- 支持通过 `search_str` 参数进行全局搜索
- 支持分页查询，默认每页显示20条
- 建议提供模糊搜索功能，提升用户体验
3. **用户交互流程**
- 第一步：获取并显示基础数据（任务类型、状态、人员）
- 第二步：用户搜索并选择关联需求
- 第三步：（可选）上传截图，获取图片 URL
- 第四步：填写任务信息并创建
4. **参数校验**
- 必填参数不能为空
- 时间格式必须为 `YYYY-MM-DD HH:mm:ss`
- 紧急程度必须在 1-5 之间
- 工时必须为正整数
5. **错误处理**
- API 调用失败时，显示明确的错误信息
- 网络异常时提供重试机制
- 参数错误时提示具体字段
6. **日志记录**
- 记录所有 API 调用日志
- 记录任务创建成功/失败信息
- 记录错误堆栈，便于问题排查
## 规范文档
- 代码规范: `Docs/PRD/01规范文档/_PRD_规范文档_代码规范.md`
- 问题总结: `Docs/PRD/01规范文档/_PRD_问题总结_记录文档.md`
- 方法总结: `Docs/PRD/01规范文档/_PRD_方法总结_记录文档.md`
- 文档规范: `Docs/PRD/01规范文档/_PRD_规范文档_文档规范.md`
- 测试规范: `Docs/PRD/01规范文档/_PRD_规范文档_测试规范.md`
---
Docs/PRD/接口安全测试/需求文档/_PRD_对接ERP任务创建_需求文档.md 0 → 100644
浏览文件 @ 387ab3eb
# ERP创建任务对接需求文档
## 项目背景
### 现有系统
本项目基于现有的**新统一平台接口安全测试系统**进行扩展，现有系统位于：
- **代码路径**：`AuxiliaryTool/ScriptTool/ApiSecurityTest/`
- **主入口**：`run_all.py`
- **现有功能**：
1. 执行OWASP API Security Top 10全量安全测试（10个模块、90个用例）
2. 历史漏洞回归测试（35个历史漏洞）
3. 华为安全红线合规检查（27项）
4. 生成Markdown格式安全测试报告
5. 自动上传报告至网盘（`\\192.168.9.9\deploy\18其它系统\安全测试\04新统一平台-安全报告\`）
### 业务场景
当前系统在完成安全测试后，报告自动上传到网盘，但缺少后续的任务跟踪机制。每次安全测试发现的问题需要人工在ERP系统中创建任务进行修复跟踪，流程不够自动化。
### 扩展目标
**目标：** 在现有安全测试流程中增加ERP任务创建功能，实现测试完成后自动/手动创建ERP任务，将测试结果记录到ERP系统中进行跟踪管理。
### 完整业务流程
```
┌─────────────────────────────────────────────────────┐
│ 1. 执行API安全测试（已有功能） │
│ ├── OWASP Top 10全量测试 │
│ ├── 历史漏洞回归测试 │
│ └── 华为红线检查 │
└───────────────────┬─────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 2. 生成安全测试报告（已有功能） │
│ └── 报告保存到 reports/ 目录（Markdown格式） │
└───────────────────┬─────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 3. 上传报告到网盘（已有功能） │
│ └── 上传到 \\192.168.9.9\...\安全报告\ │
└───────────────────┬─────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────┐
│ 4. 创建ERP任务（🆕 新功能） │
│ ├── 读取测试报告，提取测试摘要和漏洞统计 │
│ ├── 获取基础数据（任务类型、人员、需求等） │
│ ├── 填充任务信息（测试目标、测试结果、报告链接等） │
│ └── 调用ERP接口创建任务 │
└─────────────────────────────────────────────────────┘
```
### 集成方式
在 `run_all.py` 中已预留接口：
```python
def upload_to_erp(report_path):
"""
对接ERP创建任务跟踪（预留功能）
暂无对接，仅保留接口入口
"""
log.info("ERP对接功能暂未实现，跳过")
```
本需求将实现此函数，完成ERP任务创建功能。
---
## 代码路径
- **现有测试代码**：`AuxiliaryTool/ScriptTool/ApiSecurityTest/`
- **新增ERP模块**：`AuxiliaryTool/ScriptTool/ApiSecurityTest/utils/erp_client.py`
- **配置文件**：`AuxiliaryTool/ScriptTool/ApiSecurityTest/config.yaml`（扩展配置）
## ERP对接文档
- 需求文档路径：`Docs/PRD/接口安全测试/ERP对接/ERP任务模块_API接口文档.md`
## 功能需求
### 功能目标
**目标：** 对接ERP任务管理接口，在安全测试完成后自动创建ERP任务进行记录和跟踪，支持设置任务基本信息、指派人员、关联需求、上传截图等功能。
### 需求描述
#### 1. 获取基础数据
在创建任务前，需要获取以下基础数据供用户选择：
- **获取任务类型**
- 接口：`GET /openclaw/tasktype`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/tasktype`
- X-Api-Key: 保持原有的key传入
- 返回数据示例：
```json
{
"success": 1,
"data": [
{"id": 1, "name": "开发任务"},
{"id": 2, "name": "测试任务"},
{"id": 3, "name": "设计任务"},
{"id": 4, "name": "部署任务"},
{"id": 5, "name": "文档任务"},
{"id": 6, "name": "维护任务"},
{"id": 7, "name": "研究任务"},
{"id": 8, "name": "会议沟通"},
{"id": 9, "name": "售前支持"},
{"id": 10, "name": "售后支持"}
]
}
```
- 用途：用户选择任务类型
- **获取任务状态**
- 接口：`GET /openclaw/taskstatus`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/taskstatus`
- X-Api-Key: 保持原有的key传入
- 返回数据示例：
```json
{
"success": 1,
"data": [
{"id": 1, "name": "新建"},
{"id": 2, "name": "进行中"},
{"id": 3, "name": "已解决"},
{"id": 4, "name": "完成确认中"},
{"id": 10, "name": "已完成"}
]
}
```
- 用途：用户选择任务初始状态（通常选择"新建"）
- **获取人员列表**
- 接口：`GET /openclaw/stuff`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/stuff`
- X-Api-Key: 保持原有的key传入
- 返回数据示例：
```json
{
"success": 1,
"data": [
{"id": 1, "name": "超管员", "department_id": null},
{"id": 20, "name": "郑晓兵", "department_id": 94020884},
{"id": 24, "name": "黄史恭", "department_id": 596543063},
{"id": 31, "name": "欧阳友平", "department_id": 363531662}
]
}
```
- 用途：用户选择任务责任人（user_id）和下单人员（createuser_id）
- **获取需求列表**
- 接口：`GET /openclaw/develop`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/develop`
- X-Api-Key: 保持原有的key传入
- 查询参数：
- `pageNum`: 页码，默认 1
- `pageSize`: 每页数量，默认 10
- `search_str`: 传入`【新统一平台】公司内部安全测试扫描及修复验证`
- 返回数据示例：
```json
{
"success": 1,
"pageSize": 10,
"pageNum": 1,
"totalRows": 183,
"data": [
{
"id": 215,
"name": "测试需求",
"project_id": 1702,
"project_name": "演示测试项目2",
"dutyuser_id": 20,
"dutyuser_name": "郑晓兵"
}
]
}
```
- 用途：用户选择要关联的需求（develop_id 为必填）
#### 2. 创建任务（核心功能）
- 创建任务接口：
- 接口：`POST /openclaw/task`
- 接口调用地址：`https://office.ubainsyun.com:5082/api/uerp/openclaw/task`
- X-Api-Key: 保持原有的key传入
- Content-Type: `application/json`
**必填参数（通过用户输入或选择）：**
| 参数 | 类型 | 说明 | 数据来源 |
|------|------|------|----------|
| name | string | 任务名称（最长128字符） | 用户输入 |
| type_id | int | 任务类型 ID | 从任务类型列表中选择 |
| status_id | int | 任务状态 ID | 从任务状态列表中选择（默认1-新建） |
| level | int | 紧急程度（1-5） | 用户选择或输入 |
| excepthour | int | 期望工时（小时） | 用户输入 |
| excepttime | string | 截止时间 `YYYY-MM-DD HH:mm:ss` | 用户选择或输入 |
| content | string | 任务描述（HTML 格式） | 用户输入，支持富文本 |
| user_id | int | 主责任人 ID | 从人员列表中选择 |
| createuser_id | int | 下单人员 ID（创建人） | 从人员列表中选择 |
| develop_id | int | 关联的需求 ID | 从需求列表中选择 |
**可选参数：**
| 参数 | 类型 | 说明 | 数据来源 |
|------|------|------|----------|
| planhour | int | 预计工时（小时） | 用户输入，status_id > 1 时必填 |
| starttime | string | 开始时间 `YYYY-MM-DD HH:mm:ss` | 用户选择，status_id > 1 时必填 |
| endtime | string | 结束时间 `YYYY-MM-DD HH:mm:ss` | 用户选择，status_id > 1 时必填 |
| actualhour | int | 实际工时（小时） | 用户输入，默认 0 |
| project_id | int | 关联的项目 ID | 用户选择 |
| descript | string | 备注说明 | 用户输入 |
| process | int | 当前进度（0-100） | 用户输入，默认 0 |
| sort | int | 排序值 | 用户输入，默认 0 |
| copyuserList | array | 抄送人员 ID 列表 | 从人员列表中多选 |
**紧急程度（level）对照表：**
| level | 名称 | 说明 |
|-------|------|------|
| 1 | 每日观察 | 每日跟踪进度（最低优先级） |
| 2 | 非常紧急 | 需要立即处理 |
| 3 | 紧急处理 | 优先处理 |
| 4 | 常规处理 | 正常排期处理 |
| 5 | 迭代优化 | 后续版本处理（最高优先级） |
**请求示例：**
```bash
curl -X POST "https://office.ubainsyun.com:5082/api/uerp/openclaw/task" \
-H "X-Api-Key: your_api_key" \
-H "Content-Type: application/json" \
-d '{
"name": "用户登录模块接口开发",
"type_id": 1,
"status_id": 1,
"level": 3,
"excepthour": 8,
"excepttime": "2026-06-25 18:00:00",
"content": "<h4>任务描述</h4><p>完成用户登录模块的后端接口开发</p><h4>技术要求</h4><ul><li>使用 RESTful API 规范</li><li>JWT token 认证</li><li>接口文档同步更新</li></ul><h4>参考截图</h4><p><img src=\"https://office.ubainsyun.com:5015/upload/20260617223125747.png\" alt=\"接口设计图\"/></p>",
"user_id": 20,
"createuser_id": 5,
"develop_id": 128,
"planhour": 6,
"project_id": 42,
"descript": "优先级较高的开发任务"
}'
```
**返回数据示例：**
```json
{
"success": 1,
"data": {
"create": 1024,
"name": "用户登录模块接口开发"
}
}
```
### 实现要求
1. **基础数据缓存**
- 获取任务类型、任务状态、人员列表后应进行缓存
- 避免每次创建任务都重复请求基础数据
- 建议缓存时长：1小时
2. **需求列表搜索**
- 支持通过 `search_str` 参数进行全局搜索
- 支持分页查询，默认每页显示20条
- 建议提供模糊搜索功能，提升用户体验
3. **用户交互流程**
- 第一步：获取并显示基础数据（任务类型、状态、人员）
- 第二步：用户搜索并选择关联需求
- 第三步：（可选）上传截图，获取图片 URL
- 第四步：填写任务信息并创建
4. **参数校验**
- 必填参数不能为空
- 时间格式必须为 `YYYY-MM-DD HH:mm:ss`
- 紧急程度必须在 1-5 之间
- 工时必须为正整数
5. **错误处理**
- API 调用失败时，显示明确的错误信息
- 网络异常时提供重试机制
- 参数错误时提示具体字段
6. **日志记录**
- 记录所有 API 调用日志
- 记录任务创建成功/失败信息
- 记录错误堆栈，便于问题排查
## 规范文档
- 代码规范: `Docs/PRD/01规范文档/_PRD_规范文档_代码规范.md`
- 问题总结: `Docs/PRD/01规范文档/_PRD_问题总结_记录文档.md`
- 方法总结: `Docs/PRD/01规范文档/_PRD_方法总结_记录文档.md`
- 文档规范: `Docs/PRD/01规范文档/_PRD_规范文档_文档规范.md`
- 测试规范: `Docs/PRD/01规范文档/_PRD_规范文档_测试规范.md`
---
\ No newline at end of file
Docs/PRD/接口安全测试/需求文档/_PRD_对接ERP任务创建_需求文档_计划执行.md 0 → 100644
浏览文件 @ 387ab3eb
# ERP对接任务创建功能执行计划
## 执行概述
### 项目背景
本项目是在现有**新统一平台接口安全测试系统**基础上的功能扩展，而非独立工具开发。
#### 现有系统架构
- **代码位置**：`AuxiliaryTool/ScriptTool/ApiSecurityTest/`
- **主程序**：`run_all.py`
- **核心功能**：
1. 执行OWASP API Security Top 10全量安全测试
2. 历史漏洞回归测试（35个历史漏洞）
3. 华为安全红线合规检查（27项）
4. 生成Markdown格式安全测试报告
5. 自动上传报告至网盘
#### 扩展需求
现有系统在完成安全测试并上传报告后，缺少后续的任务跟踪机制。每次安全测试发现的问题需要人工在ERP系统中创建任务进行修复跟踪，流程不够自动化。
**需求**：在现有安全测试流程中增加ERP任务创建功能，实现测试完成后自动创建ERP任务，将测试结果记录到ERP系统中进行跟踪管理。
#### 集成方式
在 `run_all.py` 第55-64行已预留接口：
```python
def upload_to_erp(report_path):
"""
对接ERP创建任务跟踪（预留功能）
暂无对接，仅保留接口入口
参数:
report_path: 报告文件路径
"""
log.info("ERP对接功能暂未实现，跳过")
```
本计划将实现此函数，完成ERP任务创建功能。
#### 完整业务流程
```
步骤1：执行API安全测试（已有）
├── OWASP Top 10全量测试（90个用例）
├── 历史漏洞回归测试（35个漏洞）
└── 华为红线检查（27项）
步骤2：生成安全测试报告（已有）
└── 报告保存到 reports/ 目录（Markdown格式）
步骤3：上传报告到网盘（已有）
└── 上传到 \\192.168.9.9\...\安全报告\
步骤4：创建ERP任务（🆕 本项目实现）
├── 读取测试报告，提取测试摘要
├── 获取ERP基础数据（缓存机制）
├── 填充任务信息（测试目标、测试结果、报告链接等）
└── 调用ERP接口创建任务
```
### 执行目标
1. **功能实现**：在现有测试系统中增加ERP任务创建模块
2. **自动化**：测试完成后自动创建ERP任务（可选手动触发）
3. **数据提取**：从测试报告中提取关键信息（漏洞统计、测试摘要等）
4. **任务信息**：自动填充任务信息，支持手动调整
5. **错误处理**：完善的错误处理和日志记录
6. **配置管理**：扩展现有配置文件，添加ERP相关配置
### 技术栈
- **开发语言**：Python 3.8+（与现有系统保持一致）
- **HTTP请求**：requests库（现有系统已使用）
- **配置管理**：YAML（扩展现有config.yaml）
- **日志记录**：logging模块（复用现有logger）
- **数据处理**：json、datetime、re、pathlib
### 代码路径
- **现有测试代码**：`AuxiliaryTool/ScriptTool/ApiSecurityTest/`
- **新增ERP模块**：`utils/erp_client.py`（ERP客户端）
- **新增任务模块**：`utils/task_creator.py`（任务创建器）
- **配置扩展**：`config.yaml`（增加ERP相关配置）
- **测试用例**：`tests/test_erp_client.py`（单元测试）
---
## 任务分解与实施计划
### 阶段一：模块设计与配置（预计0.5天）
#### 1.1 目录结构设计
在现有安全测试项目中新增ERP模块：
```
AuxiliaryTool/ScriptTool/ApiSecurityTest/
├── utils/
│ ├── erp_client.py # 🆕 ERP客户端（API调用）
│ ├── task_creator.py # 🆕 任务创建器（业务逻辑）
│ └── report_parser.py # 🆕 报告解析器（提取测试结果）
├── tests/
│ └── test_erp_client.py # 🆕 ERP客户端单元测试
├── config.yaml # 📝 扩展配置（添加ERP相关配置）
└── run_all.py # 📝 修改（实现upload_to_erp函数）
```
#### 1.2 配置文件扩展
- [ ] 扩展`config.yaml`，添加ERP相关配置
```yaml
# ERP配置
erp:
enabled: true # 是否启用ERP对接
base_url: "https://office.ubainsyun.com:5082/api/uerp"
api_key: "your_api_key_here" # ERP API密钥
timeout: 30 # 请求超时（秒）
retry_interval: 2 # 重试间隔（秒）
max_retries: 3 # 最大重试次数
# 任务配置
task:
type_id: 2 # 默认任务类型：测试任务
status_id: 1 # 默认任务状态：新建
level: 3 # 默认紧急程度：紧急处理
excepthour: 8 # 默认期望工时（小时）
user_id: 20 # 默认责任人ID
createuser_id: 5 # 默认创建人ID
# 需求关联
develop_search_str: "【新统一平台】公司内部安全测试扫描及修复验证"
# 缓存配置
cache_duration: 3600 # 基础数据缓存时长（秒）
# 上传配置
upload:
enabled: true # 是否上传截图
max_images: 5 # 最多上传图片数量
```
- [ ] 配置项说明
- 所有ERP相关配置集中在`erp`节点下
- 支持启用/禁用开关（`enabled`）
- 任务配置支持默认值，减少手动输入
- 缓存配置避免重复请求基础数据
- 使用说明
- 配置说明
- [ ] 创建`run.py`启动脚本
#### 1.2 配置管理实现
- [ ] 创建`config.py`配置模块
- ERP API配置（base_url、api_key）
- 超时配置（timeout、retry_interval、max_retries）
- 缓存配置（cache_duration）
- 日志配置（log_level、log_file）
- [ ] 创建`config.json`配置文件
```json
{
"erp": {
"base_url": "https://office.ubainsyun.com:5082/api/uerp",
"api_key": "your_api_key_here",
"timeout": 30,
"retry_interval": 2,
"max_retries": 3
},
"cache": {
"enabled": true,
"duration": 3600,
"file": "cache_data.json"
},
"log": {
"level": "INFO",
"file": "logs/erp_task_creation.log",
"max_size": 10485760,
"backup_count": 5
}
}
```
### 阶段二：ERP客户端开发（预计1天）
#### 2.1 ERP客户端实现
- [ ] 创建`utils/erp_client.py`ERP客户端类
```python
"""
ERP客户端模块
对接ERP任务管理接口，提供任务创建相关功能
"""
import requests
import time
from typing import Optional, Dict, List
from utils.logger import log
class ERPClient:
"""ERP系统客户端"""
def __init__(self, config):
"""
初始化客户端
参数:
config: 配置字典（从config.yaml读取）
"""
self.base_url = config['erp']['base_url']
self.api_key = config['erp']['api_key']
self.timeout = config['erp'].get('timeout', 30)
self.retry_interval = config['erp'].get('retry_interval', 2)
self.max_retries = config['erp'].get('max_retries', 3)
def get_task_types(self) -> Optional[List[Dict]]:
"""获取任务类型列表"""
def get_task_statuses(self) -> Optional[List[Dict]]:
"""获取任务状态列表"""
def get_staff_list(self) -> Optional[List[Dict]]:
"""获取人员列表"""
def get_develop_list(self, page_num=1, page_size=20, search_str="") -> Optional[Dict]:
"""
获取需求列表（支持搜索和分页）
参数:
page_num: 页码
page_size: 每页数量
search_str: 搜索字符串
返回:
包含data、totalRows等字段的字典
"""
def create_task(self, task_data: Dict) -> Optional[Dict]:
"""
创建任务
参数:
task_data: 任务数据字典
返回:
创建结果（包含任务ID）
"""
def _make_request(self, method: str, endpoint: str, **kwargs) -> Optional[Dict]:
"""
统一的请求方法（含重试机制）
参数:
method: HTTP方法（GET/POST）
endpoint: API端点
**kwargs: 其他请求参数
返回:
响应数据字典
"""
```
- [ ] 实现请求重试机制
- 网络超时重试
- 服务端错误（5xx）重试
- 客户端错误（4xx）不重试
- 固定间隔重试（retry_interval秒）
- [ ] API端点定义（在模块顶部定义常量）
```python
# API端点
TASK_TYPE_ENDPOINT = "/openclaw/tasktype"
TASK_STATUS_ENDPOINT = "/openclaw/taskstatus"
STAFF_ENDPOINT = "/openclaw/stuff"
DEVELOP_ENDPOINT = "/openclaw/develop"
TASK_ENDPOINT = "/openclaw/task"
UPLOAD_IMAGE_ENDPOINT = "/openclaw/upload/richtext"
```
#### 2.2 数据模型定义
- [ ] 在`erp_client.py`中定义数据类（使用dataclass或简单类）
```python
@dataclass
class TaskType:
id: int
name: str
@dataclass
class TaskStatus:
id: int
name: str
@dataclass
class Staff:
id: int
name: str
department_id: Optional[int]
@dataclass
class DevelopRequirement:
id: int
name: str
project_id: int
project_name: str
dutyuser_id: int
dutyuser_name: str
```
```python
@dataclass
class TaskType:
id: int
name: str
@dataclass
class TaskStatus:
id: int
name: str
@dataclass
class Staff:
id: int
name: str
department_id: Optional[int]
@dataclass
class DevelopRequirement:
id: int
name: str
project_id: int
project_name: str
dutyuser_id: int
dutyuser_name: str
@dataclass
class TaskCreateRequest:
name: str
type_id: int
status_id: int
level: int
excepthour: int
excepttime: str
content: str
user_id: int
createuser_id: int
develop_id: int
# 可选字段
planhour: Optional[int] = None
starttime: Optional[str] = None
endtime: Optional[str] = None
actualhour: Optional[int] = None
project_id: Optional[int] = None
descript: Optional[str] = None
process: Optional[int] = None
sort: Optional[int] = None
copyuserList: Optional[List[int]] = None
```
### 阶段三：报告解析与任务创建（预计1天）
#### 3.1 报告解析器
- [ ] 创建`utils/report_parser.py`报告解析模块
```python
"""
报告解析器模块
从Markdown格式的安全测试报告中提取关键信息
"""
import re
from typing import Dict, Optional
from utils.logger import log
class ReportParser:
"""安全测试报告解析器"""
def __init__(self, report_path: str):
"""
初始化解析器
参数:
report_path: 报告文件路径
"""
self.report_path = report_path
self.content = None
def load(self) -> bool:
"""加载报告文件"""
def parse_summary(self) -> Dict:
"""
解析测试摘要
返回:
包含以下字段的字典：
- total: 总测试用例数
- high: 高危漏洞数
- medium: 中危漏洞数
- low: 低危漏洞数
- info: 信息类数
- safe: 安全项数
- target: 测试目标URL
- start_time: 测试开始时间
- end_time: 测试结束时间
"""
def parse_vulnerabilities(self) -> list:
"""
解析漏洞列表
返回:
漏洞列表，每个漏洞包含：
- id: 漏洞编号
- name: 漏洞名称
- level: 风险等级
- description: 漏洞描述
"""
def get_high_risk_list(self) -> list:
"""获取高危漏洞列表"""
def generate_task_name(self) -> str:
"""
生成ERP任务名称
格式：{目标IP}接口安全测试及漏洞修复_{日期}
示例：192.168.5.44接口安全测试及漏洞修复_20260624
"""
def generate_task_content(self) -> str:
"""
生成ERP任务描述（HTML格式）
包含：
- 测试目标
- 测试时间
- 漏洞统计（表格形式）
- 高危漏洞列表
- 报告链接
"""
```
- [ ] 实现正则表达式解析
- 解析测试摘要（使用正则匹配漏洞数量）
- 解析漏洞详情
- 提取测试时间和目标
- [ ] 实现HTML内容生成
- 生成表格形式的漏洞统计
- 高危漏洞列表突出显示
- 添加报告网盘链接
#### 3.2 任务创建器
- [ ] 创建`utils/task_creator.py`任务创建模块
```python
"""
任务创建器模块
协调ERP客户端、报告解析器、缓存管理，完成任务创建
"""
from datetime import datetime, timedelta
from utils.erp_client import ERPClient
from utils.report_parser import ReportParser
from utils.logger import log
class TaskCreator:
"""任务创建器"""
def __init__(self, erp_client: ERPClient, config: Dict):
"""
初始化任务创建器
参数:
erp_client: ERP客户端实例
config: 配置字典
"""
self.erp_client = erp_client
self.config = config
self.cache = {}
def create_task_from_report(self, report_path: str) -> Optional[Dict]:
"""
从测试报告创建ERP任务
参数:
report_path: 报告文件路径
返回:
创建的任务信息（包含任务ID），失败返回None
"""
def _get_basic_data(self) -> bool:
"""
获取基础数据（使用缓存）
从ERP获取任务类型、状态、人员等基础数据
缓存在内存中，避免重复请求
"""
def _search_requirement(self) -> Optional[Dict]:
"""
搜索关联的需求
使用配置的搜索字符串查找需求
"""
def _build_task_data(self, report: ReportParser) -> Dict:
"""
构建任务数据
从配置和报告中提取数据，填充任务字段
"""
def _calculate_deadline(self) -> str:
"""
计算截止时间
默认：当前时间 + 3天
"""
```
- [ ] 实现缓存机制
- 内存缓存基础数据
- 缓存过期检查（config中的cache_duration）
- [ ] 实现任务数据构建
- 从配置读取默认值
- 从报告中提取测试结果
- 生成任务描述HTML
### 阶段四：系统集成（预计0.5天）
#### 4.1 修改run_all.py
- [ ] 实现`upload_to_erp`函数
```python
def upload_to_erp(report_path):
"""
对接ERP创建任务跟踪
参数:
report_path: 报告文件路径
返回:
bool: 是否成功创建任务
"""
# 1. 检查是否启用ERP对接
if not config.get('erp', {}).get('enabled', False):
log.info("ERP对接未启用，跳过任务创建")
return False
# 2. 初始化ERP客户端
erp_client = ERPClient(config)
# 3. 创建任务创建器
task_creator = TaskCreator(erp_client, config)
# 4. 从报告创建任务
try:
result = task_creator.create_task_from_report(report_path)
if result:
log.info(f"✓ ERP任务创建成功：任务ID={result['create']}, 名称={result['name']}")
return True
else:
log.error("✗ ERP任务创建失败")
return False
except Exception as e:
log.error(f"✗ ERP任务创建异常：{e}")
return False
```
- [ ] 在测试流程中调用
- 在`run_all_tests()`函数的最后阶段调用
- 在`run_single_module()`函数中也调用
- [ ] 添加错误处理
- ERP创建失败不影响报告上传
- 记录详细日志便于排查
### 阶段五：测试与优化（预计0.5天）
#### 5.1 单元测试
- [ ] 创建`tests/test_erp_client.py`
- 测试ERP客户端各方法
- Mock API响应
- [ ] 测试报告解析器
- 使用真实报告测试
- 验证解析结果正确性
#### 5.2 集成测试
- [ ] 完整流程测试
- 执行安全测试
- 生成报告
- 创建ERP任务
- 验证任务正确创建
#### 5.3 日志优化
- [ ] 增强日志输出
- API调用详细日志
- 任务创建进度提示
- 成功/失败明确标识
"""初始化基础数据步骤"""
def create_ui(self):
"""创建UI界面"""
def fetch_data(self):
"""获取基础数据（任务类型、状态、人员）"""
def display_data(self):
"""显示获取到的数据"""
def on_next(self):
"""下一步按钮处理"""
```
- [ ] 实现数据显示
- 任务类型下拉列表
- 任务状态下拉列表
- 责任人下拉列表
- 创建人下拉列表
- [ ] 实现缓存机制
- 首次请求API
- 后续从缓存读取
- 缓存过期自动刷新
#### 4.3 第二步：需求选择
- [ ] 创建`gui/step2_requirement.py`
```python
class RequirementStep:
def __init__(self, parent, api_client):
"""初始化需求选择步骤"""
def create_ui(self):
"""创建UI界面"""
def on_search(self):
"""搜索按钮处理"""
def display_results(self, requirements):
"""显示搜索结果（分页）"""
def on_select(self):
"""选择需求处理"""
def on_prev_page(self):
"""上一页"""
def on_next_page(self):
"""下一页"""
```
- [ ] 实现搜索功能
- 搜索输入框
- 默认搜索字符串："【新统一平台】公司内部安全测试扫描及修复验证"
- 支持自定义搜索
- [ ] 实现分页功能
- 默认每页20条
- 显示总记录数和当前页码
- 上一页/下一页按钮
- [ ] 实现需求选择
- 单选按钮或双击选择
- 显示需求详细信息（ID、名称、项目、负责人）
#### 4.4 第三步：截图上传（可选）
- [ ] 创建`gui/step3_upload.py`
```python
class UploadStep:
def __init__(self, parent, api_client):
"""初始化上传步骤"""
def create_ui(self):
"""创建UI界面"""
def on_select_files(self):
"""选择文件按钮处理"""
def on_upload(self):
"""上传按钮处理"""
def on_clear(self):
"""清空按钮处理"""
def skip_step(self):
"""跳过此步骤"""
```
- [ ] 实现文件选择
- 支持多图片选择
- 支持的格式：PNG、JPG、JPEG
- 显示预览缩略图
- [ ] 实现图片上传
- 调用图片上传接口
- 获取图片URL列表
- 显示上传进度
- [ ] 实现跳过功能
- 允许用户跳过此步骤
- 不上传截图直接创建任务
#### 4.5 第四步：任务创建
- [ ] 创建`gui/step4_create.py`
```python
class TaskCreateStep:
def __init__(self, parent, api_client, validator):
"""初始化任务创建步骤"""
def create_ui(self):
"""创建UI界面"""
def load_basic_data(self):
"""加载基础数据（从步骤1）"""
def load_requirement(self):
"""加载选中的需求（从步骤2）"""
def load_images(self):
"""加载上传的图片URL（从步骤3）"""
def on_create(self):
"""创建任务按钮处理"""
def validate_inputs(self):
"""校验用户输入"""
def build_task_data(self):
"""构建任务数据"""
def generate_content_html(self):
"""生成任务描述HTML"""
```
- [ ] 实现表单输入
- 任务名称（文本输入，最长128字符）
- 任务类型（下拉，从步骤1获取）
- 任务状态（下拉，从步骤1获取，默认"新建"）
- 紧急程度（下拉1-5，默认3-紧急处理）
- 期望工时（数字输入，小时）
- 截止时间（日期时间选择器）
- 任务描述（富文本编辑器或简单文本框）
- 主责任人（下拉，从步骤1获取）
- 下单人员（下拉，从步骤1获取）
- 关联需求（只读，从步骤2获取）
- 关联项目（可选下拉）
- 抄送人员（多选下拉列表）
- 预计工时（可选，status_id>1时必填）
- 开始时间（可选，status_id>1时必填）
- 结束时间（可选，status_id>1时必填）
- 实际工时（可选，默认0）
- 备注说明（可选文本框）
- 当前进度（可选，0-100，默认0）
- [ ] 实现HTML内容生成
- 任务描述标题
- 任务描述正文
- 参考截图（从步骤3的URL）
```html
<h4>任务描述</h4>
<p>{任务描述内容}</p>
<h4>参考截图</h4>
<p><img src="{图片URL1}" alt="截图1"/></p>
<p><img src="{图片URL2}" alt="截图2"/></p>
```
- [ ] 实现参数校验
- 必填字段检查
- 字段长度检查
- 数据格式检查
- 实时校验提示
- [ ] 实现任务创建
- 显示确认对话框
- 调用创建任务API
- 显示创建结果
- 成功后返回第一步
```
---
## 验收标准
### 功能验收
#### 1. 报告解析 ✓
- [ ] 能够正确加载Markdown报告文件
- [ ] 能够提取测试摘要（总用例数、各等级漏洞数）
- [ ] 能够提取测试时间（开始/结束）
- [ ] 能够提取测试目标URL
- [ ] 能够生成规范的HTML任务描述
- [ ] 任务名称格式正确（含IP地址和日期）
#### 2. ERP接口调用 ✓
- [ ] 能够成功获取任务类型列表（10种类型）
- [ ] 能够成功获取任务状态列表（5种状态）
- [ ] 能够成功获取人员列表
- [ ] 能够成功搜索需求列表
- [ ] 能够成功调用创建任务接口
- [ ] 数据缓存功能正常（1小时内不重复请求）
#### 3. 自动化流程 ✓
- [ ] 测试完成后自动触发任务创建
- [ ] 配置禁用时能够跳过ERP对接
- [ ] ERP创建失败不影响报告上传
- [ ] 任务ID正确记录和显示
#### 4. 错误处理 ✓
- [ ] 网络超时有重试机制（最多3次）
- [ ] API错误有明确日志记录
- [ ] 报告解析失败有错误提示
- [ ] 配置错误有明确提示
- [ ] 日志记录完整（请求/响应/错误）
### 性能验收
#### 1. 响应时间
- [ ] 报告解析时间 < 1秒
- [ ] 基础数据获取时间 < 3秒
- [ ] 需求搜索响应时间 < 2秒
- [ ] 任务创建响应时间 < 5秒
- [ ] 总体任务创建耗时 < 15秒
#### 2. 稳定性
- [ ] 连续测试10次无崩溃
- [ ] 网络波动后能够正常恢复
- [ ] 长时间运行无内存泄漏
- [ ] ERP服务异常时不影响测试报告生成
### 集成验收
#### 1. 端到端流程 ✓
- [ ] 完整执行安全测试流程
- [ ] 测试报告正确生成
- [ ] 报告上传网盘成功
- [ ] ERP任务创建成功
- [ ] 任务信息正确填充
#### 2. 任务信息验证
- [ ] 任务名称格式正确
- [ ] 任务描述包含测试摘要
- [ ] 漏洞统计信息准确
- [ ] 关联需求ID正确
- [ ] 责任人和创建人正确
- [ ] 截止时间合理（默认+3天）
- [ ] 界面布局合理美观
- [ ] 字体大小适中
- [ ] 颜色搭配协调
- [ ] 图标含义清晰
#### 2. 操作便捷
- [ ] 步骤引导清晰
- [ ] 必填项有明确标识
- [ ] 错误提示准确及时
- [ ] 支持键盘快捷键
#### 3. 信息完整
- [ ] 操作有进度提示
- [ ] 结果有明确反馈
- [ ] 帮助文档完善
---
## 测试计划
### 单元测试
#### 1. API模块测试
- [ ] 测试ERPClient的各个方法
- `get_task_types()`
- `get_task_statuses()`
- `get_staff_list()`
- `get_develop_list()`
- `create_task()`
- [ ] 测试重试机制
- [ ] 测试错误处理
#### 2. 工具模块测试
- [ ] 测试CacheManager
- 缓存写入
- 缓存读取
- 缓存过期
- 缓存清空
- [ ] 测试TaskValidator
- 各字段校验规则
- 边界值测试
- 异常值测试
#### 3. 数据模型测试
- [ ] 测试各数据类的序列化/反序列化
- [ ] 测试数据类的验证逻辑
### 集成测试
#### 1. 端到端测试
- [ ] 完整任务创建流程测试
1. 启动程序
2. 获取基础数据
3. 搜索并选择需求
4. 上传截图（可选）
5. 填写任务信息
6. 创建任务
7. 验证结果
#### 2. 异常场景测试
- [ ] 网络断开时的行为
- [ ] API返回错误时的行为
- [ ] 参数错误时的行为
- [ ] 缓存过期时的行为
### 用户验收测试
#### 1. 功能测试
- [ ] 按照用户手册进行完整操作
- [ ] 验证所有功能点可用
- [ ] 验证结果符合预期
#### 2. 易用性测试
- [ ] 新用户首次使用
- [ ] 老用户日常使用
- [ ] 收集使用反馈
---
## 风险评估
### 技术风险
#### 1. API接口变更风险 🔴 高
- **风险描述**：ERP系统接口可能发生变更，导致功能异常
- **影响范围**：所有功能
- **缓解措施**：
- 接口版本管理
- 充分的错误处理
- 及时关注接口变更通知
- 定期进行接口测试
#### 2. 网络稳定性风险 🟡 中
- **风险描述**：网络不稳定导致请求失败
- **影响范围**：所有API调用
- **缓解措施**：
- 实现重试机制
- 提供离线缓存
- 友好的错误提示
- 网络状态检测
#### 3. 数据一致性风险 🟡 中
- **风险描述**：基础数据更新后缓存未及时刷新
- **影响范围**：任务类型、状态、人员列表
- **缓解措施**：
- 合理设置缓存时长
- 提供手动刷新按钮
- 缓存失效时自动更新
### 项目风险
#### 1. 时间延期风险 🟡 中
- **风险描述**：GUI开发复杂度超出预期
- **影响范围**：项目整体进度
- **缓解措施**：
- 分阶段开发
- 优先实现核心功能
- 界面美化可后续优化
#### 2. 资源不足风险 🟢 低
- **风险描述**：开发资源投入不足
- **影响范围**：功能完整性
- **缓解措施**：
- 合理排期
- 及时沟通风险
- 必要时调整范围
---
## 实施记录
### 进度记录
| 阶段 | 计划时间 | 实际时间 | 完成度 | 备注 |
|------|---------|---------|-------|------|
| 阶段一：项目架构搭建 | 1天 | ___天 | [ ] % | 待开始 |
| 阶段二：API接口模块开发 | 1-2天 | ___天 | [ ] % | 待开始 |
| 阶段三：工具模块开发 | 1天 | ___天 | [ ] % | 待开始 |
| 阶段四：系统集成 | 0.5天 | ___天 | [ ] % | 待开始 |
| 阶段五：测试与优化 | 0.5天 | ___天 | [ ] % | 待开始 |
### 里程碑
| 里程碑 | 预计日期 | 实际日期 | 状态 | 备注 |
|--------|---------|---------|------|------|
| 项目启动 | YYYY-MM-DD | YYYY-MM-DD | [ ] 待完成 | - |
| ERP客户端完成 | YYYY-MM-DD | YYYY-MM-DD | [ ] 待完成 | - |
| 报告解析器完成 | YYYY-MM-DD | YYYY-MM-DD | [ ] 待完成 | - |
| 集成测试完成 | YYYY-MM-DD | YYYY-MM-DD | [ ] 待完成 | - |
| 项目交付 | YYYY-MM-DD | YYYY-MM-DD | [ ] 待完成 | - |
---
## 后续工作
### 功能扩展
- [ ] 支持手动触发任务创建（独立脚本）
- [ ] 支持自定义任务模板（不同测试类型）
- [ ] 支持任务进度同步（从ERP读取任务状态）
- [ ] 支持批量历史报告导入（批量创建任务）
- [ ] 支持任务列表查看和导出
### 监控优化
- [ ] 定时任务监控（检测长时间未更新的任务）
- [ ] 任务提醒功能（截止日期提醒）
- [ ] 周报统计（本周创建任务数量和完成情况）
- [ ] 操作历史记录
- [ ] 常用配置保存
- [ ] 智能填充（基于历史数据）
### 技术优化
- [ ] 数据库支持（SQLite）
- [ ] 配置文件加密（API Key）
- [ ] 自动更新功能
- [ ] 多语言支持（i18n）
- [ ] 命令行模式支持
---
## 附录
### A. 参考资料
1. **项目文档**
- 代码规范：`Docs/PRD/01规范文档/_PRD_规范文档_代码规范.md`
- 文档规范：`Docs/PRD/01规范文档/_PRD_规范文档_文档规范.md`
- 测试规范：`Docs/PRD/01规范文档/_PRD_规范文档_测试规范.md`
- 问题总结：`Docs/PRD/01规范文档/_PRD_问题总结_记录文档.md`
- 方法总结：`Docs/PRD/01规范文档/_PRD_方法总结_记录文档.md`
2. **相关工具**
- ERP对接示例：`AuxiliaryTool/FunctionalTestReportGeneration/src/erp_uploader.py`
- 安全测试主程序：`AuxiliaryTool/ScriptTool/ApiSecurityTest/run_all.py`
- 报告生成器：`AuxiliaryTool/ScriptTool/ApiSecurityTest/utils/report_generator.py`
### B. 相关接口文档
| 接口 | 方法 | 路径 | 说明 |
|------|------|------|------|
| 获取任务类型 | GET | /openclaw/tasktype | 获取所有任务类型 |
| 获取任务状态 | GET | /openclaw/taskstatus | 获取所有任务状态 |
| 获取人员列表 | GET | /openclaw/stuff | 获取所有人员 |
| 获取需求列表 | GET | /openclaw/develop | 获取需求列表（支持搜索） |
| 创建任务 | POST | /openclaw/task | 创建新任务 |
| 上传图片 | POST | /openclaw/upload/richtext | 上传任务截图 |
### C. 紧急程度对照表
| level | 名称 | 说明 | 使用场景 |
|-------|------|------|---------|
| 1 | 每日观察 | 每日跟踪进度（最低优先级） | 长期优化类任务 |
| 2 | 非常紧急 | 需要立即处理 | 生产故障、紧急bug |
| 3 | 紧急处理 | 优先处理 | 重要功能开发 |
| 4 | 常规处理 | 正常排期处理 | 常规开发任务 |
| 5 | 迭代优化 | 后续版本处理（最高优先级） | 性能优化、体验改进 |
### D. 开发环境要求
- **Python版本**：3.8或更高（与现有系统保持一致）
- **操作系统**：Windows 10/11、Linux（EulerOS）
- **依赖库**：
```txt
requests>=2.28.0
pyyaml>=6.0
```
- **现有依赖**（已安装）：
```txt
colorama>=0.4.6
```
- **IDE推荐**：PyCharm、VSCode
---
## 优化功能回填
- [ ] 支持手动触发模式（独立脚本，不依赖安全测试流程）
- [ ] 添加任务模板支持（针对不同测试类型使用不同模板）
- [ ] 实现任务进度回写（从ERP读取任务状态更新到本地）
- [ ] 支持批量历史报告处理（扫描历史报告批量创建任务）
- [ ] 添加任务通知功能（钉钉/企微通知新任务创建）
- [ ] 实现任务统计周报（本周创建任务、完成任务统计）
- [ ] 支持任务优先级自动调整（根据漏洞数量自动设置紧急程度）
- [ ] 添加任务关联功能（同一需求多次测试关联到同一任务）
---
**文档版本**：v2.0
**创建日期**：2026-06-24
**最后更新**：2026-06-24
**文档状态**：✅ 已完成
**负责人**：开发团队
**重要变更说明**：
- v2.0为重大架构调整
- 从独立GUI工具调整为现有安全测试系统的扩展模块
- 采用自动化流程，测试完成后自动创建ERP任务
- 无需GUI界面，通过配置文件控制行为
---
## 变更记录
| 版本 | 日期 | 变更内容 | 变更人 |
|------|------|---------|--------|
| v2.0 | 2026-06-24 | 重大架构调整：从独立GUI工具改为现有系统扩展模块 | 开发团队 |
| v1.0 | 2026-06-24 | 初始版本创建（已废弃） | 开发团队 |
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论