central-task/.trae/documents/任务中心管理系统实现计划.md

任务中心管理系统最终实现计划

项目结构设计

1. 创建Django项目：task_center
2. 创建任务管理应用：tasks
3. 配置数据库为SQLite
4. 集成Bootstrap前端框架
5. 配置pytest测试框架

数据库模型设计

Client模型

字段名	类型	描述
id	AutoField	客户端ID
name	CharField(unique=True)	客户端标识
token	CharField(max_length=128)	API Token
last_seen	DateTimeField	最后活跃时间
created_at	DateTimeField	创建时间

Task模型

字段名	类型	描述
id	AutoField	任务ID
name	CharField	任务名称
client_name	CharField(null=True, blank=True)	指定执行客户端
script	TextField(null=True, blank=True)	执行脚本
status	CharField(choices=STATUS_CHOICES, default='pending')	任务状态
timeout_seconds	IntegerField(default=259200)	超时时间（默认3天=259200秒）
created_at	DateTimeField	创建时间
updated_at	DateTimeField	更新时间
assigned_to	CharField(null=True, blank=True)	实际执行客户端
started_at	DateTimeField(null=True, blank=True)	开始执行时间
completed_at	DateTimeField(null=True, blank=True)	完成时间

TaskResult模型

字段名	类型	描述
id	AutoField	结果ID
task	ForeignKey(Task)	关联任务
client	ForeignKey(Client)	执行客户端
result_file	FileField(upload_to='task_results/')	结果文件
status	CharField(choices=STATUS_CHOICES)	执行状态
message	TextField(null=True, blank=True)	执行消息
created_at	DateTimeField	创建时间

状态定义

STATUS_CHOICES = [
    ('pending', '待分配'),
    ('assigned', '已分配'),
    ('running', '执行中'),
    ('success', '成功'),
    ('failed', '失败'),
    ('retrying', '重试中'),
    ('timeout', '超时，关闭'),
]

API接口设计

认证机制

• 所有API端点均需token认证

• 客户端通过HTTP头Authorization: Token <token>进行身份验证

• 使用Django REST Framework的TokenAuthentication

• 未提供有效token的请求将返回401 Unauthorized

任务管理API

• GET /api/tasks/ - 获取任务列表（需认证）

• GET /api/tasks/<id>/ - 获取任务详情（需认证）

• POST /api/tasks/ - 创建任务（需认证）

• PUT /api/tasks/<id>/ - 更新任务（需认证）

• DELETE /api/tasks/<id>/ - 删除任务（需认证）

客户端API

• POST /api/tasks/claim/ - 客户端原子认领任务（需认证）

• POST /api/tasks/<id>/start/ - 客户端开始执行任务（需认证）

• POST /api/tasks/<id>/complete/ - 客户端完成任务（需认证）

• POST /api/task_results/ - 上传任务结果（需认证，支持文件上传）

客户端管理API

• GET /api/clients/ - 获取客户端列表（需认证）

• POST /api/clients/ - 创建客户端（需认证）

• GET /api/clients/<id>/ - 获取客户端详情（需认证）

文件下载API

• GET /api/task_results/<id>/download/ - 下载任务结果文件（需认证）

文件上传实现细节

1. Django媒体文件配置

# settings.py
MEDIA_URL = '/media/'
MEDIA_ROOT = os.path.join(BASE_DIR, 'media')

2. 文件上传API实现

• 使用Django REST Framework的MultiPartParser和FormParser

• API端点：POST /api/task_results/

• 支持multipart/form-data格式上传文件

• 上传字段：task_id、status、message、result_file

3. 前端文件上传实现

• 使用HTML5的input type="file"元素

• 表单设置enctype="multipart/form-data"

• 使用Bootstrap样式美化文件上传控件

• 支持进度条显示（可选）

4. 文件存储策略

• 本地文件系统存储：media/task_results/目录

• 文件名自动生成，避免冲突

• 支持大文件上传（通过Django默认配置）

5. 文件下载实现

• API端点：GET /api/task_results/<id>/download/

• 返回Content-Disposition: attachment头，触发浏览器下载

• 支持断点续传（通过Django默认配置）

前端页面设计

1. 任务列表页：展示所有任务，支持筛选和搜索
2. 任务创建页：表单创建新任务
3. 任务详情页：查看任务详情和执行结果历史
4. 客户端管理页：管理客户端列表
5. 结果文件上传页：支持文件上传和状态更新
6. 结果文件下载功能：点击即可下载

核心功能实现

1. 全API Token认证

• 所有API视图均使用TokenAuthentication

• 配置DEFAULT_AUTHENTICATION_CLASSES和DEFAULT_PERMISSION_CLASSES

2. 原子任务认领机制

• 使用数据库事务确保任务认领的原子性

• 客户端调用/api/tasks/claim/时，系统自动查找并分配可用任务

3. 任务超时自动回收

• 使用Django管理命令定期检查超时任务

• 超过timeout_seconds的任务自动设置为timeout状态

• 管理命令：python manage.py check_task_timeouts

4. 任务结果版本管理

• TaskResult模型记录每次执行结果

• 支持查看任务的完整执行历史

• 支持下载不同版本的结果文件

测试用例设计

目录结构

task_center/
├── tasks/
│   ├── tests/
│   │   ├── __init__.py
│   │   ├── test_models.py
│   │   ├── test_views.py
│   │   ├── test_api.py
│   │   ├── test_integration.py
│   │   └── test_factories.py
│   └── ...
├── conftest.py
└── pytest.ini

测试类型

1. 模型测试：测试模型字段、方法和关系
2. API测试：测试所有API端点的功能和认证机制
3. 文件上传测试：测试文件上传和下载功能
4. 视图测试：测试前端视图渲染
5. 集成测试：测试完整的任务流程
6. 工厂测试：使用factory_boy创建测试数据

实现步骤

1. 创建Django项目和应用
2. 配置项目设置（数据库、REST Framework、认证、媒体文件等）
3. 实现数据库模型
4. 实现API视图和序列化器，包括文件上传功能
5. 配置URL路由
6. 实现前端模板，包括文件上传表单
7. 实现任务超时管理命令
8. 编写测试用例
9. 测试API接口和功能

技术栈

• 后端：Django 5.0.6 + Django REST Framework

• 前端：HTML + Bootstrap 5

• 数据库：SQLite

• 文件存储：本地文件系统

• 测试：pytest + factory_boy

预期效果

• 所有API端点均需要token认证

• 支持结果文件的上传和下载功能

• 后台可通过admin或API创建和管理任务

• 客户端通过API进行身份验证，原子认领任务

• 支持任务超时自动回收

• 完整的任务执行历史记录

• 简单大方的Bootstrap前端界面

• 全面的测试用例覆盖