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

from typing import Any

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,
)
from app.schemas.task_center import (
    ListProjectTasksResponse,
    ListTasksResponse,
    ProjectTaskResponse,
    UserTaskResponse,
)
from fastapi import APIRouter, Depends, HTTPException

from packages.application import (
    CreateGenerationTaskCommand,
    CreateGenerationTaskUseCase,
    SubmitIngestJobCommand,
    SubmitIngestJobUseCase,
)

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 _status_value(status) -> str:
    """安全获取状态值（兼容 StrEnum 和 plain string）。"""
    return status.value if hasattr(status, "value") else str(status)


def _generation_step(task) -> str:
    s = _status_value(task.status)
    if s == "pending":
        return "等待 Worker 执行"
    if s == "running":
        return "正在生成成片"
    if s == "completed":
        return "生成完成"
    if s == "failed":
        return "生成失败"
    return s


def _ingest_step(job) -> str:
    s = _status_value(job.status)
    if s == "pending":
        return "等待导入"
    if s == "processing":
        return "正在分析素材"
    if s == "completed":
        return "导入完成"
    if s == "failed":
        return "导入失败"
    return s


def _generation_task_to_project_response(task) -> ProjectTaskResponse:
    return ProjectTaskResponse(
        id=f"generation:{task.id}",
        task_type="generation",
        project_id=task.project_id,
        status=_status_value(task.status),
        progress=task.progress,
        current_step=_generation_step(task),
        error_message=task.error_message,
        user_message=_humanize_task_error(task.error_message),
        retryable=_status_value(task.status) == "failed",
        source_id=task.id,
        template_id=task.template_id,
        created_at=task.created_at,
        updated_at=task.completed_at or task.started_at or task.created_at,
    )


# ── 用户级端点（放在项目级端点之前，避免路由冲突） ──


@router.get("/tasks", response_model=ListTasksResponse)
def list_user_tasks(
    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),
) -> ListTasksResponse:
    """用户级任务列表（跨 project），合并 ingest + generation 任务。"""
    user_id = authenticated_user.user.id
    items: list[UserTaskResponse] = []

    for task in generation_task_repository.list_by_user(user_id):
        items.append(
            UserTaskResponse(
                id=f"generation:{task.id}",
                task_type="generation",
                project_id=task.project_id,
                template_id=task.template_id,
                status=_status_value(task.status),
                progress=task.progress,
                current_step=_generation_step(task),
                error_message=task.error_message,
                user_message=_humanize_task_error(task.error_message),
                retryable=_status_value(task.status) == "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 ListTasksResponse(items=items)


@router.post("/tasks/{task_id}/retry", response_model=UserTaskResponse)
def retry_task_by_id(
    task_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    generation_task_repository: Any = Depends(get_generation_task_repository),
) -> UserTaskResponse:
    """简化重试：通过 task_id 直接重试失败的生成任务。"""
    task = generation_task_repository.get(task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Generation task not found")
    if task.created_by_user_id and task.created_by_user_id != authenticated_user.user.id:
        raise HTTPException(status_code=403, detail="Access denied to this task")
    if _status_value(task.status) != "failed":
        raise HTTPException(status_code=409, detail="Only failed tasks can be retried")

    use_case = CreateGenerationTaskUseCase(generation_task_repository)
    retried = use_case.execute(
        CreateGenerationTaskCommand(
            project_id=task.project_id,
            asset_library_id=task.asset_library_id,
            strategy_id=task.strategy_id,
            voice_library_id=task.voice_library_id,
            template_id=task.template_id,
            asset_ids=task.asset_ids,
            title_ids=task.title_ids,
            voice_ids=task.voice_ids,
            created_by_user_id=authenticated_user.user.id,
        )
    )
    celery_app.send_task("worker.generate_video", args=[retried.id])
    return UserTaskResponse(
        id=f"generation:{retried.id}",
        task_type="generation",
        project_id=retried.project_id,
        template_id=retried.template_id,
        status=_status_value(retried.status),
        progress=retried.progress,
        current_step=_generation_step(retried),
        source_id=retried.id,
        created_at=retried.created_at,
        updated_at=retried.created_at,
    )


# ── 项目级端点 ──


@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),
) -> ListProjectTasksResponse:
    project = project_repository.find_by_id(project_id)
    if project is None:
        raise HTTPException(status_code=404, detail="Project not found")

    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",
                project_id=job.project_id,
                status=_status_value(job.status),
                progress=100.0 if _status_value(job.status) == "completed" else 0.0,
                current_step=_ingest_step(job),
                error_message=job.error_message,
                user_message=_humanize_task_error(job.error_message),
                retryable=_status_value(job.status) == "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(_generation_task_to_project_response(task))
    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),
) -> 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")
        if _status_value(task.status) != "failed":
            raise HTTPException(status_code=409, detail="Only failed tasks can be retried")
        use_case = CreateGenerationTaskUseCase(generation_task_repository)
        retried = use_case.execute(
            CreateGenerationTaskCommand(
                project_id=task.project_id,
                asset_library_id=task.asset_library_id,
                strategy_id=task.strategy_id,
                voice_library_id=task.voice_library_id,
                template_id=task.template_id,
                asset_ids=task.asset_ids,
                title_ids=task.title_ids,
                voice_ids=task.voice_ids,
                created_by_user_id=authenticated_user.user.id,
            )
        )
        celery_app.send_task("worker.generate_video", args=[retried.id])
        return _generation_task_to_project_response(retried)
    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")
        if _status_value(job.status) != "failed":
            raise HTTPException(status_code=409, detail="Only failed tasks can be retried")
        use_case = SubmitIngestJobUseCase(ingest_job_repository)
        retried = use_case.execute(
            SubmitIngestJobCommand(
                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",
            project_id=retried.project_id,
            status=_status_value(retried.status),
            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")