feat(tasks): add project task center

apps/api/app/api/router.py

@@ -10,6 +10,7 @@ from app.api.routes.ingest_jobs import router as ingest_jobs_router
 from app.api.routes.project_management import router as project_management_router
 from app.api.routes.project_titles import router as project_titles_router
 from app.api.routes.projects import router as projects_router
+from app.api.routes.task_center import router as task_center_router
 from app.api.routes.upload import router as upload_router
 from app.api.routes.workspaces import router as workspaces_router
 from fastapi import APIRouter
@@ -35,6 +36,10 @@ api_router.include_router(
     project_titles_router,
     tags=["标题库"],
 )
+api_router.include_router(
+    task_center_router,
+    tags=["任务中心"],
+)
 api_router.include_router(
     asset_diagnosis_router,
     tags=["素材诊断"],

apps/api/app/api/routes/task_center.py

@@ -0,0 +1,105 @@
+from typing import Any
+
+from app.api.routes.permissions import require_workspace_member
+from app.auth import AuthenticatedUser, get_current_user
+from app.dependencies import (
+    get_generation_task_repository,
+    get_ingest_job_repository,
+    get_project_repository,
+    get_workspace_member_repository,
+)
+from app.schemas.task_center import ListProjectTasksResponse, ProjectTaskResponse
+from fastapi import APIRouter, Depends, HTTPException
+
+from packages.ports.workspace_member_repository import WorkspaceMemberRepository
+
+router = APIRouter()
+
+
+def _humanize_task_error(error_message: str) -> str:
+    raw = (error_message or "").strip()
+    if not raw:
+        return ""
+    lower = raw.lower()
+    if "ffmpeg" in lower or "ffprobe" in lower or "invalid data" in lower or "moov atom" in lower:
+        return "视频素材格式无法识别，请重新导出为常见 MP4/H.264 后再试。"
+    if "oss" in lower or "bucket" in lower or "storage" in lower:
+        return "素材存储服务读取或写入失败，请稍后重试或联系小虾检查 OSS。"
+    if "not found" in lower or "no such file" in lower:
+        return "任务依赖的素材或文件不存在，请确认素材仍在项目中。"
+    return f"任务失败：{raw}"
+
+
+def _generation_step(task) -> str:
+    if task.status.value == "pending":
+        return "等待 Worker 执行"
+    if task.status.value == "running":
+        return "正在生成成片"
+    if task.status.value == "completed":
+        return "生成完成"
+    if task.status.value == "failed":
+        return "生成失败"
+    return task.status.value
+
+
+def _ingest_step(job) -> str:
+    if job.status.value == "pending":
+        return "等待导入"
+    if job.status.value == "processing":
+        return "正在分析素材"
+    if job.status.value == "completed":
+        return "导入完成"
+    if job.status.value == "failed":
+        return "导入失败"
+    return job.status.value
+
+
+@router.get("/projects/{project_id}/tasks", response_model=ListProjectTasksResponse)
+def list_project_tasks(
+    project_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+    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),
+) -> ListProjectTasksResponse:
+    project = project_repository.find_by_id(project_id)
+    if project is None:
+        raise HTTPException(status_code=404, detail="Project not found")
+    require_workspace_member(project.workspace_id, authenticated_user, workspace_member_repository)
+
+    items: list[ProjectTaskResponse] = []
+    for job in ingest_job_repository.list_by_project(project_id):
+        items.append(ProjectTaskResponse(
+            id=f"ingest:{job.id}",
+            task_type="ingest",
+            workspace_id=job.workspace_id,
+            project_id=job.project_id,
+            status=job.status.value,
+            progress=100.0 if job.status.value == "completed" else 0.0,
+            current_step=_ingest_step(job),
+            error_message=job.error_message,
+            user_message=_humanize_task_error(job.error_message),
+            retryable=job.status.value == "failed",
+            source_id=job.id,
+            created_at=job.created_at,
+            updated_at=job.updated_at,
+        ))
+    for task in generation_task_repository.list_by_project(project_id):
+        items.append(ProjectTaskResponse(
+            id=f"generation:{task.id}",
+            task_type="generation",
+            workspace_id=task.workspace_id,
+            project_id=task.project_id,
+            status=task.status.value,
+            progress=task.progress,
+            current_step=_generation_step(task),
+            error_message=task.error_message,
+            user_message=_humanize_task_error(task.error_message),
+            retryable=task.status.value == "failed",
+            source_id=task.id,
+            created_at=task.created_at,
+            updated_at=task.completed_at or task.started_at or task.created_at,
+        ))
+    items.sort(key=lambda item: item.updated_at or item.created_at or "", reverse=True)
+    return ListProjectTasksResponse(items=items)

apps/api/app/schemas/task_center.py

@@ -0,0 +1,22 @@
+from datetime import datetime
+from pydantic import BaseModel, Field
+
+
+class ProjectTaskResponse(BaseModel):
+    id: str
+    task_type: str
+    workspace_id: str
+    project_id: str
+    status: str
+    progress: float
+    current_step: str
+    error_message: str = ""
+    user_message: str = ""
+    retryable: bool = False
+    source_id: str = ""
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+class ListProjectTasksResponse(BaseModel):
+    items: list[ProjectTaskResponse] = Field(default_factory=list)

apps/web/e2e/core-generation.spec.ts

@@ -188,5 +188,11 @@ test.describe('Core generation and download flow', () => {
     expect(titleAfterGeneration.status(), await titleAfterGeneration.text()).toBe(200);
     const titlesData = (await titleAfterGeneration.json()) as { items: ProjectTitleResponse[] };
     expect(titlesData.items.find((item) => item.id === titleData.id)?.usage_count).toBe(1);
+
+    await page.goto(`/projects/${projectData.id}/tasks`);
+    await expect(page.getByText('项目任务中心')).toBeVisible({ timeout: 20_000 });
+    await expect(page.getByText('视频生成')).toBeVisible({ timeout: 20_000 });
+    await expect(page.getByText('已完成')).toBeVisible({ timeout: 20_000 });
+    await expect(page.getByText(createdTask.id)).toBeVisible({ timeout: 20_000 });
   });
 });

apps/web/src/api/tasks.ts

@@ -0,0 +1,22 @@
+import apiClient from './client';
+
+export interface ProjectTaskItem {
+  id: string;
+  task_type: 'ingest' | 'generation' | string;
+  workspace_id: string;
+  project_id: string;
+  status: string;
+  progress: number;
+  current_step: string;
+  error_message: string;
+  user_message: string;
+  retryable: boolean;
+  source_id: string;
+  created_at?: string | null;
+  updated_at?: string | null;
+}
+
+export const getProjectTasks = async (projectId: string): Promise<ProjectTaskItem[]> => {
+  const response = await apiClient.get(`/projects/${projectId}/tasks`);
+  return response.data.items;
+};

apps/web/src/pages/workspace/ProjectTasks.tsx

@@ -0,0 +1,82 @@
+import React from 'react';
+import { Alert, Button, Card, Empty, List, Progress, Space, Tag, Typography } from 'antd';
+import { useQuery } from '@tanstack/react-query';
+import { useParams } from 'react-router-dom';
+import { getProjectTasks } from '@/api/tasks';
+
+const statusMap: Record<string, { label: string; color: string }> = {
+  pending: { label: '排队中', color: 'default' },
+  processing: { label: '处理中', color: 'blue' },
+  running: { label: '运行中', color: 'blue' },
+  completed: { label: '已完成', color: 'green' },
+  failed: { label: '失败', color: 'red' },
+  cancelled: { label: '已取消', color: 'default' },
+};
+
+const taskTypeLabels: Record<string, string> = {
+  ingest: '素材导入',
+  generation: '视频生成',
+};
+
+const ProjectTasks: React.FC = () => {
+  const { id } = useParams<{ id: string }>();
+  const projectId = id || '';
+
+  const tasksQuery = useQuery({
+    queryKey: ['project-tasks', projectId],
+    queryFn: () => getProjectTasks(projectId),
+    enabled: !!projectId,
+    refetchInterval: 5000,
+  });
+
+  return (
+    <div style={{ padding: 24 }}>
+      <Card title="项目任务中心" extra={<Button onClick={() => tasksQuery.refetch()}>刷新</Button>}>
+        {tasksQuery.isError && <Alert style={{ marginBottom: 16 }} type="error" showIcon message="任务列表加载失败" />}
+        {tasksQuery.data?.some((task) => task.retryable) && (
+          <Alert
+            style={{ marginBottom: 16 }}
+            type="warning"
+            showIcon
+            message="部分任务可重试"
+            description="重试入口将在任务执行器安全幂等化后开放；当前先展示失败原因，避免假重试。"
+          />
+        )}
+        <List
+          loading={tasksQuery.isLoading}
+          dataSource={tasksQuery.data || []}
+          locale={{ emptyText: <Empty description="暂无任务，上传素材或发起生成后会出现在这里" /> }}
+          renderItem={(task) => {
+            const status = statusMap[task.status] || { label: task.status, color: 'default' };
+            return (
+              <List.Item>
+                <List.Item.Meta
+                  title={
+                    <Space wrap>
+                      <Typography.Text>{taskTypeLabels[task.task_type] || task.task_type}</Typography.Text>
+                      <Tag color={status.color}>{status.label}</Tag>
+                      <Tag>{task.current_step}</Tag>
+                      {task.retryable ? <Tag color="orange">可重试（暂未开放）</Tag> : null}
+                    </Space>
+                  }
+                  description={
+                    <Space direction="vertical" style={{ width: '100%' }}>
+                      <Progress percent={Math.round(task.progress || 0)} size="small" status={task.status === 'failed' ? 'exception' : undefined} />
+                      {task.user_message ? <Typography.Text type="danger">{task.user_message}</Typography.Text> : null}
+                      <Typography.Text type="secondary" copyable>
+                        任务 ID：{task.source_id}
+                      </Typography.Text>
+                    </Space>
+                  }
+                />
+              </List.Item>
+            );
+          }}
+        />
+      </Card>
+    </div>
+  );
+};
+
+export const Component = ProjectTasks;
+export default ProjectTasks;

apps/web/src/pages/workspace/WorkspaceDetail.tsx

@@ -100,6 +100,9 @@ const WorkspaceDetail: React.FC = () => {
                     <Button key="generation" type="link" onClick={() => navigate(`/projects/${project.id}/generation`, { state: { workspaceId: id } })}>
                       视频生成
                     </Button>,
+                    <Button key="tasks" type="link" onClick={() => navigate(`/projects/${project.id}/tasks`, { state: { workspaceId: id } })}>
+                      任务中心
+                    </Button>,
                     <Button key="results" type="link" onClick={() => navigate(`/projects/${project.id}/results`, { state: { workspaceId: id } })}>
                       生成结果
                     </Button>,

apps/web/src/router/index.tsx

@@ -72,6 +72,10 @@ export const router = createBrowserRouter([
         path: 'projects/:id/generation',
         lazy: () => import('@/pages/workspace/ProjectGeneration'),
       },
+      {
+        path: 'projects/:id/tasks',
+        lazy: () => import('@/pages/workspace/ProjectTasks'),
+      },
       {
         path: 'projects/:id/results',
         lazy: () => import('@/pages/workspace/ProjectResults'),

packages/adapters/sqlalchemy_impl/ingest_job_repository.py

@@ -42,6 +42,10 @@ class SQLAlchemyIngestJobRepository:
             updated_at=model.updated_at,
         )
 
+    def list_by_project(self, project_id: str) -> list[IngestJob]:
+        models = self.session.query(IngestJobModel).filter(IngestJobModel.project_id == project_id).all()
+        return [self.get(model.id) for model in models if self.get(model.id) is not None]
+
     def update(self, job: IngestJob) -> IngestJob:
         model = self.session.query(IngestJobModel).filter(IngestJobModel.id == job.id).first()
         if model is None:

tests/unit/test_task_center.py

@@ -0,0 +1,25 @@
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from app.api.routes.task_center import _humanize_task_error, _ingest_step
+from packages.domain import IngestJob, IngestJobStatus
+
+
+def test_humanize_task_error_for_media_failures():
+    assert "MP4" in _humanize_task_error("ffmpeg invalid data found")
+    assert "OSS" in _humanize_task_error("oss bucket read failed")
+    assert "不存在" in _humanize_task_error("file not found")
+
+
+def test_ingest_step_is_user_readable():
+    job = IngestJob.create(
+        workspace_id="workspace-1",
+        project_id="project-1",
+        library_id="library-1",
+        storage_key="uploads/video.mp4",
+    )
+    job.status = IngestJobStatus.COMPLETED
+
+    assert _ingest_step(job) == "导入完成"