xiaoxia-saas/apps/api/app/api/routes/task_center.py

from typing import Any

from app.api.routes.permissions import require_workspace_member
from app.auth import AuthenticatedUser, get_current_user
from app.core.celery_app import celery_app
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.application import (
    CreateGenerationTaskCommand,
    CreateGenerationTaskUseCase,
    SubmitIngestJobCommand,
    SubmitIngestJobUseCase,
)
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)


@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,
                edit_plan_id=task.edit_plan_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")