Audit Bot
ea9e536740
方案A: 后端扩展3处API改动，支持前端模板中心化模型。
1. POST /api/v1/generation/tasks — project_id改为可选，新增template_id/asset_ids/title_ids/voice_ids
2. GET /api/v1/generation/tasks + GET /api/v1/tasks — 新增用户级列表（跨project）
3. POST /api/v1/tasks/{task_id}/retry + POST /api/v1/generation/tasks/{task_id}/retry — 简化重试
改动涉及：
- Domain: GenerationTask新增6个字段，放宽create()校验
- Application: CreateGenerationTaskCommand新增字段
- Ports: GenerationTaskRepository新增list_by_user()
- Adapter: SQLAlchemy模型+仓储实现新字段和list_by_user
- Schema: 双模式校验（project模式/模板模式）
- Routes: generation_tasks + task_center双路由注册
- Migration: 015_add_generation_task_extensions
向后兼容：所有旧端点和参数不变。
57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from uuid import uuid4
|
|
|
|
from packages.domain import GenerationTask
|
|
from packages.ports.generation_task_repository import GenerationTaskRepository
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class CreateGenerationTaskCommand:
|
|
project_id: str = ""
|
|
asset_library_id: str = ""
|
|
strategy_id: str = ""
|
|
voice_library_id: str = ""
|
|
edit_plan_id: str = ""
|
|
template_id: str = ""
|
|
asset_ids: list[str] = field(default_factory=list)
|
|
title_ids: list[str] = field(default_factory=list)
|
|
voice_ids: list[str] = field(default_factory=list)
|
|
created_by_user_id: str = ""
|
|
|
|
|
|
class CreateGenerationTaskUseCase:
|
|
def __init__(self, generation_task_repository: GenerationTaskRepository):
|
|
self.generation_task_repository = generation_task_repository
|
|
|
|
def execute(self, command: CreateGenerationTaskCommand) -> GenerationTask:
|
|
task = GenerationTask(
|
|
id=uuid4().hex,
|
|
project_id=command.project_id,
|
|
asset_library_id=command.asset_library_id,
|
|
strategy_id=command.strategy_id,
|
|
voice_library_id=command.voice_library_id,
|
|
edit_plan_id=command.edit_plan_id,
|
|
template_id=command.template_id,
|
|
asset_ids=command.asset_ids,
|
|
title_ids=command.title_ids,
|
|
voice_ids=command.voice_ids,
|
|
status="pending",
|
|
progress=0.0,
|
|
result_count=0,
|
|
error_message="",
|
|
started_at=None,
|
|
completed_at=None,
|
|
created_by_user_id=command.created_by_user_id,
|
|
)
|
|
return self.generation_task_repository.create(task)
|
|
|
|
|
|
class GetGenerationTaskUseCase:
|
|
def __init__(self, generation_task_repository: GenerationTaskRepository):
|
|
self.generation_task_repository = generation_task_repository
|
|
|
|
def execute(self, task_id: str) -> GenerationTask | None:
|
|
return self.generation_task_repository.get(task_id)
|