diff --git a/apps/api/app/api/routes/task_center.py b/apps/api/app/api/routes/task_center.py index d821acc0b..647d30fc1 100644 --- a/apps/api/app/api/routes/task_center.py +++ b/apps/api/app/api/routes/task_center.py @@ -9,8 +9,10 @@ from app.dependencies import ( get_workspace_member_repository, ) from app.schemas.task_center import ListProjectTasksResponse, ProjectTaskResponse +from app.core.celery_app import celery_app from fastapi import APIRouter, Depends, HTTPException +from packages.application import CreateGenerationTaskCommand, CreateGenerationTaskUseCase, SubmitIngestJobCommand, SubmitIngestJobUseCase from packages.ports.workspace_member_repository import WorkspaceMemberRepository router = APIRouter() @@ -103,3 +105,60 @@ def list_project_tasks( )) items.sort(key=lambda item: item.updated_at or item.created_at or "", reverse=True) return ListProjectTasksResponse(items=items) + + +@router.post("/tasks/{task_type}/{source_id}/retry", response_model=ProjectTaskResponse) +def retry_project_task( + task_type: str, + source_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + ingest_job_repository: Any = Depends(get_ingest_job_repository), + generation_task_repository: Any = Depends(get_generation_task_repository), + workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository), +) -> ProjectTaskResponse: + if task_type == "generation": + task = generation_task_repository.get(source_id) + if task is None: + raise HTTPException(status_code=404, detail="Generation task not found") + require_workspace_member(task.workspace_id, authenticated_user, workspace_member_repository) + if task.status.value != "failed": + raise HTTPException(status_code=409, detail="Only failed tasks can be retried") + use_case = CreateGenerationTaskUseCase(generation_task_repository) + retried = use_case.execute(CreateGenerationTaskCommand( + workspace_id=task.workspace_id, + project_id=task.project_id, + asset_library_id=task.asset_library_id, + strategy_id=task.strategy_id, + voice_library_id=task.voice_library_id, + created_by_user_id=authenticated_user.user.id, + )) + celery_app.send_task("worker.generate_video", args=[retried.id]) + return ProjectTaskResponse( + id=f"generation:{retried.id}", task_type="generation", workspace_id=retried.workspace_id, + project_id=retried.project_id, status=retried.status.value, progress=retried.progress, + current_step=_generation_step(retried), source_id=retried.id, created_at=retried.created_at, + updated_at=retried.created_at, + ) + if task_type == "ingest": + job = ingest_job_repository.get(source_id) + if job is None: + raise HTTPException(status_code=404, detail="Ingest job not found") + require_workspace_member(job.workspace_id, authenticated_user, workspace_member_repository) + if job.status.value != "failed": + raise HTTPException(status_code=409, detail="Only failed tasks can be retried") + use_case = SubmitIngestJobUseCase(ingest_job_repository) + retried = use_case.execute(SubmitIngestJobCommand( + workspace_id=job.workspace_id, + project_id=job.project_id, + library_id=job.library_id, + storage_key=job.storage_key, + )) + celery_app.send_task("worker.ingest_asset", args=[retried.id]) + return ProjectTaskResponse( + id=f"ingest:{retried.id}", task_type="ingest", workspace_id=retried.workspace_id, + project_id=retried.project_id, status=retried.status.value, progress=0, + current_step=_ingest_step(retried), source_id=retried.id, created_at=retried.created_at, + updated_at=retried.updated_at, + ) + raise HTTPException(status_code=400, detail="Unsupported task type") + diff --git a/apps/web/src/api/tasks.ts b/apps/web/src/api/tasks.ts index fc8fe8ff4..22a67b9d2 100644 --- a/apps/web/src/api/tasks.ts +++ b/apps/web/src/api/tasks.ts @@ -20,3 +20,8 @@ export const getProjectTasks = async (projectId: string): Promise => { + const response = await apiClient.post(`/tasks/${taskType}/${sourceId}/retry`); + return response.data; +}; diff --git a/apps/web/src/pages/workspace/ProjectTasks.tsx b/apps/web/src/pages/workspace/ProjectTasks.tsx index 384d30cf0..86a41de62 100644 --- a/apps/web/src/pages/workspace/ProjectTasks.tsx +++ b/apps/web/src/pages/workspace/ProjectTasks.tsx @@ -1,8 +1,8 @@ import React from 'react'; import { Alert, Button, Card, Empty, List, Progress, Space, Tag, Typography } from 'antd'; -import { useQuery } from '@tanstack/react-query'; +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; import { useParams } from 'react-router-dom'; -import { getProjectTasks } from '@/api/tasks'; +import { getProjectTasks, retryProjectTask } from '@/api/tasks'; const statusMap: Record = { pending: { label: '排队中', color: 'default' }, @@ -21,6 +21,7 @@ const taskTypeLabels: Record = { const ProjectTasks: React.FC = () => { const { id } = useParams<{ id: string }>(); const projectId = id || ''; + const queryClient = useQueryClient(); const tasksQuery = useQuery({ queryKey: ['project-tasks', projectId], @@ -29,6 +30,11 @@ const ProjectTasks: React.FC = () => { refetchInterval: 5000, }); + const retryMutation = useMutation({ + mutationFn: ({ taskType, sourceId }: { taskType: string; sourceId: string }) => retryProjectTask(taskType, sourceId), + onSuccess: () => queryClient.invalidateQueries({ queryKey: ['project-tasks', projectId] }), + }); + return (
tasksQuery.refetch()}>刷新}> @@ -39,7 +45,7 @@ const ProjectTasks: React.FC = () => { type="warning" showIcon message="部分任务可重试" - description="重试入口将在任务执行器安全幂等化后开放;当前先展示失败原因,避免假重试。" + description="失败任务可通过“重试”重新排队;系统会保留原始失败原因方便排查。" /> )} { {taskTypeLabels[task.task_type] || task.task_type} {status.label} {task.current_step} - {task.retryable ? 可重试(暂未开放) : null} + {task.retryable ? ( + + ) : null} } description={