From 43366f290ca715208f80c5027fbe7ca58544d44e Mon Sep 17 00:00:00 2001 From: CI Test Date: Wed, 1 Jul 2026 22:54:21 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E4=BB=BB?= =?UTF-8?q?=E5=8A=A12.10=20JobService=20=E2=80=94=20=E5=BC=82=E6=AD=A5?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E7=AE=A1=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 Job 领域模型（状态机、JobType/JobStatus 枚举） - 新增 JobRepository 端口 + SQLAlchemy 实现 - 新增 10 个 Use Cases（CreateJob/Submit/Progress/Complete/Fail/Retry/Cancel/Get/List/Statistics） - 新增 JobService 服务层，集成 VideoComposeService - 新增 Pydantic schemas + RESTful API 路由 - 新增 Celery compose_video 任务（含进度追踪） - 新增数据库迁移 018（jobs 表） - 新增 44 个单元测试，全部通过 - 修复状态转换：允许 pending→success（快速完成场景） --- alembic/versions/018_add_jobs_table.py | 54 ++ apps/api/app/api/router.py | 5 + apps/api/app/api/routes/jobs.py | 323 ++++++++++ apps/api/app/dependencies.py | 9 + apps/api/app/schemas/job.py | 109 ++++ apps/api/app/services/__init__.py | 2 + apps/api/app/services/job_service.py | 268 ++++++++ apps/worker/worker_app/celery_app.py | 1 + apps/worker/worker_app/tasks/__init__.py | 5 + apps/worker/worker_app/tasks/compose_video.py | 136 +++++ .../sqlalchemy_impl/job_repository.py | 169 +++++ packages/adapters/sqlalchemy_impl/models.py | 24 + packages/application/__init__.py | 30 + packages/application/jobs.py | 257 ++++++++ packages/domain/__init__.py | 4 + packages/domain/job.py | 289 +++++++++ packages/ports/job_repository.py | 64 ++ tests/unit/test_job_service.py | 578 ++++++++++++++++++ 18 files changed, 2327 insertions(+) create mode 100755 alembic/versions/018_add_jobs_table.py mode change 100644 => 100755 apps/api/app/api/router.py create mode 100755 apps/api/app/api/routes/jobs.py mode change 100644 => 100755 apps/api/app/dependencies.py create mode 100755 apps/api/app/schemas/job.py mode change 100644 => 100755 apps/api/app/services/__init__.py create mode 100755 apps/api/app/services/job_service.py mode change 100644 => 100755 apps/worker/worker_app/celery_app.py mode change 100644 => 100755 apps/worker/worker_app/tasks/__init__.py create mode 100755 apps/worker/worker_app/tasks/compose_video.py create mode 100755 packages/adapters/sqlalchemy_impl/job_repository.py mode change 100644 => 100755 packages/application/__init__.py create mode 100755 packages/application/jobs.py mode change 100644 => 100755 packages/domain/__init__.py create mode 100755 packages/domain/job.py create mode 100755 packages/ports/job_repository.py create mode 100755 tests/unit/test_job_service.py diff --git a/alembic/versions/018_add_jobs_table.py b/alembic/versions/018_add_jobs_table.py new file mode 100755 index 000000000..6a58b14a8 --- /dev/null +++ b/alembic/versions/018_add_jobs_table.py @@ -0,0 +1,54 @@ +"""Phase 8 任务 2.10: Create jobs table for unified async task management + +Revision ID: 018 +Revises: 017 +Create Date: 2026-07-01 + +新增 jobs 表，用于统一管理异步任务（视频合成、渲染等）的生命周期。 +""" + +from alembic import op +import sqlalchemy as sa + +revision = "018" +down_revision = "017" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "jobs", + sa.Column("id", sa.String(32), primary_key=True), + sa.Column("project_id", sa.String(32), nullable=False, index=True), + sa.Column("job_type", sa.String(30), nullable=False, index=True), + sa.Column("status", sa.String(20), nullable=False, server_default="pending", index=True), + sa.Column("progress", sa.Float, nullable=False, server_default="0.0"), + sa.Column("current_stage", sa.String(200), nullable=False, server_default=""), + sa.Column("payload", sa.JSON, nullable=False, server_default="{}"), + sa.Column("result", sa.JSON, nullable=False, server_default="{}"), + sa.Column("error_message", sa.Text, nullable=False, server_default=""), + sa.Column("retry_count", sa.Integer, nullable=False, server_default="0"), + sa.Column("max_retries", sa.Integer, nullable=False, server_default="3"), + sa.Column("celery_task_id", sa.String(100), nullable=False, server_default=""), + sa.Column("source_id", sa.String(32), nullable=False, server_default="", index=True), + sa.Column("created_by_user_id", sa.String(32), nullable=False, server_default="", index=True), + sa.Column("started_at", sa.DateTime, nullable=True), + sa.Column("completed_at", sa.DateTime, nullable=True), + sa.Column( + "created_at", + sa.DateTime, + nullable=False, + server_default=sa.func.now(), + ), + sa.Column( + "updated_at", + sa.DateTime, + nullable=False, + server_default=sa.func.now(), + ), + ) + + +def downgrade() -> None: + op.drop_table("jobs") diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py old mode 100644 new mode 100755 index e6fc3e70f..6a121b4c5 --- a/apps/api/app/api/router.py +++ b/apps/api/app/api/router.py @@ -11,6 +11,7 @@ from app.api.routes.edit_templates import router as edit_templates_router from app.api.routes.generated_videos import router as generated_videos_router from app.api.routes.generation_tasks import router as generation_tasks_router from app.api.routes.health import router as health_check_router +from app.api.routes.jobs import router as jobs_router from app.api.routes.ingest_jobs import router as ingest_jobs_router from app.api.routes.projects import router as projects_router from app.api.routes.recipes import router as recipes_router @@ -78,6 +79,10 @@ api_router.include_router( prefix="/generation", tags=["Generation"], ) +api_router.include_router( + jobs_router, + tags=["Job"], +) api_router.include_router( generated_videos_router, prefix="/generated-videos", diff --git a/apps/api/app/api/routes/jobs.py b/apps/api/app/api/routes/jobs.py new file mode 100755 index 000000000..ee8554ab4 --- /dev/null +++ b/apps/api/app/api/routes/jobs.py @@ -0,0 +1,323 @@ +"""Job API 路由 — Phase 8 任务 2.10. + +提供统一异步任务管理 RESTful 接口： +- POST /api/v1/jobs 创建任务 +- GET /api/v1/jobs/{job_id} 任务详情 +- GET /api/v1/projects/{project_id}/jobs 项目任务列表 +- GET /api/v1/projects/{project_id}/jobs/stats 任务统计 +- PUT /api/v1/jobs/{job_id}/progress 更新进度 +- POST /api/v1/jobs/{job_id}/complete 标记完成 +- POST /api/v1/jobs/{job_id}/fail 标记失败 +- POST /api/v1/jobs/{job_id}/retry 重试任务 +- POST /api/v1/jobs/{job_id}/cancel 取消任务 +- POST /api/v1/jobs/{job_id}/submit 提交执行 +""" + +from __future__ import annotations + +import logging +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_db_session, get_job_repository, get_project_repository +from app.schemas.job import ( + CompleteJobRequest, + CreateJobRequest, + FailJobRequest, + JobResponse, + JobStatisticsResponse, + ListJobsResponse, + UpdateProgressRequest, + job_to_response, +) +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy.orm import Session + +from packages.application.jobs import ( + CancelJobUseCase, + CompleteJobCommand, + CompleteJobUseCase, + CreateJobCommand, + CreateJobUseCase, + FailJobCommand, + FailJobUseCase, + GetJobStatisticsUseCase, + GetJobUseCase, + ListJobsUseCase, + RetryJobUseCase, + SubmitJobUseCase, + UpdateJobProgressCommand, + UpdateJobProgressUseCase, +) +from packages.domain.job import JobType + +logger = logging.getLogger(__name__) + +router = APIRouter() + +# 任务类型 → Celery task name 映射 +_JOB_TYPE_TO_CELERY_TASK: dict[str, str] = { + JobType.VIDEO_COMPOSE: "worker.compose_video", + JobType.RENDER_EDIT_PLAN: "worker.render_edit_plan", + JobType.ASSET_INGEST: "worker.ingest_asset", + JobType.CLASSIFICATION: "worker.classify_asset", + JobType.VOICE_EXTRACTION: "worker.extract_voice", + JobType.GENERATION: "worker.generate_video", +} + + +def _check_project_access(project_id: str, user_id: str, project_repository) -> None: + """检查用户是否有项目访问权限。""" + project = project_repository.find_by_id(project_id) + if project is None: + raise HTTPException(status_code=404, detail=f"Project {project_id} not found") + if not project.can_access(user_id): + raise HTTPException(status_code=403, detail="Access denied to project") + + +# ── 创建任务 ────────────────────────────────────────────────────────────────── + + +@router.post("/jobs", response_model=JobResponse, status_code=status.HTTP_201_CREATED) +def create_job( + request: CreateJobRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), + project_repository: Any = Depends(get_project_repository), +) -> JobResponse: + """创建异步任务。 + + 创建后任务处于 pending 状态，需要调用 /submit 提交执行。 + """ + _check_project_access(request.project_id, authenticated_user.user.id, project_repository) + + # 校验 job_type + try: + JobType(request.job_type) + except ValueError: + raise HTTPException( + status_code=400, + detail=f"不支持的任务类型: {request.job_type}，" + f"可选值: {[t.value for t in JobType]}", + ) + + use_case = CreateJobUseCase(job_repo) + job = use_case.execute( + CreateJobCommand( + project_id=request.project_id, + job_type=request.job_type, + payload=request.payload, + source_id=request.source_id, + created_by_user_id=authenticated_user.user.id, + max_retries=request.max_retries, + ) + ) + + return job_to_response(job) + + +# ── 提交执行 ────────────────────────────────────────────────────────────────── + + +@router.post("/jobs/{job_id}/submit", response_model=JobResponse) +def submit_job( + job_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), +) -> JobResponse: + """提交任务执行。 + + 将任务状态从 pending 切换为 running，并 dispatch Celery 异步任务。 + """ + use_case = SubmitJobUseCase(job_repo) + + try: + job = use_case.execute(job_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + # 权限检查 + if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id: + raise HTTPException(status_code=403, detail="Access denied to this job") + + # Dispatch Celery 任务 + celery_task_name = _JOB_TYPE_TO_CELERY_TASK.get(job.job_type.value) + if celery_task_name: + result = celery_app.send_task(celery_task_name, args=[job.id], kwargs=job.payload) + job.celery_task_id = result.id + job_repo.update(job) + logger.info("已提交 Celery 任务: job_id=%s celery_task_id=%s", job.id, result.id) + + return job_to_response(job) + + +# ── 查询接口 ────────────────────────────────────────────────────────────────── + + +@router.get("/jobs/{job_id}", response_model=JobResponse) +def get_job( + job_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), +) -> JobResponse: + """获取任务详情。""" + use_case = GetJobUseCase(job_repo) + job = use_case.execute(job_id) + if job is None: + raise HTTPException(status_code=404, detail=f"Job {job_id} not found") + return job_to_response(job) + + +@router.get("/projects/{project_id}/jobs", response_model=ListJobsResponse) +def list_project_jobs( + project_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), + project_repository: Any = Depends(get_project_repository), + job_type: str | None = Query(default=None, description="按任务类型过滤"), + status_filter: str | None = Query(default=None, alias="status", description="按状态过滤"), + limit: int = Query(default=50, ge=1, le=200), + offset: int = Query(default=0, ge=0), +) -> ListJobsResponse: + """获取项目下的任务列表。""" + _check_project_access(project_id, authenticated_user.user.id, project_repository) + + use_case = ListJobsUseCase(job_repo) + jobs = use_case.execute( + project_id=project_id, + job_type=job_type, + status=status_filter, + limit=limit, + offset=offset, + ) + items = [job_to_response(j) for j in jobs] + return ListJobsResponse(items=items, total=len(items)) + + +@router.get("/projects/{project_id}/jobs/stats", response_model=JobStatisticsResponse) +def get_job_statistics( + project_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), + project_repository: Any = Depends(get_project_repository), +) -> JobStatisticsResponse: + """获取项目任务统计摘要。""" + _check_project_access(project_id, authenticated_user.user.id, project_repository) + + use_case = GetJobStatisticsUseCase(job_repo) + stats = use_case.execute(project_id) + return JobStatisticsResponse(**stats) + + +# ── 进度更新 ────────────────────────────────────────────────────────────────── + + +@router.put("/jobs/{job_id}/progress", response_model=JobResponse) +def update_job_progress( + job_id: str, + request: UpdateProgressRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), +) -> JobResponse: + """更新任务进度。""" + use_case = UpdateJobProgressUseCase(job_repo) + + try: + job = use_case.execute( + UpdateJobProgressCommand( + job_id=job_id, + progress=request.progress, + current_stage=request.current_stage, + ) + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + return job_to_response(job) + + +# ── 完成 / 失败 ──────────────────────────────────────────────────────────────── + + +@router.post("/jobs/{job_id}/complete", response_model=JobResponse) +def complete_job( + job_id: str, + request: CompleteJobRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), +) -> JobResponse: + """标记任务完成。""" + use_case = CompleteJobUseCase(job_repo) + + try: + job = use_case.execute(CompleteJobCommand(job_id=job_id, result=request.result)) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + return job_to_response(job) + + +@router.post("/jobs/{job_id}/fail", response_model=JobResponse) +def fail_job( + job_id: str, + request: FailJobRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), +) -> JobResponse: + """标记任务失败。""" + use_case = FailJobUseCase(job_repo) + + try: + job = use_case.execute(FailJobCommand(job_id=job_id, error_message=request.error_message)) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + return job_to_response(job) + + +# ── 重试 / 取消 ──────────────────────────────────────────────────────────────── + + +@router.post("/jobs/{job_id}/retry", response_model=JobResponse) +def retry_job( + job_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), +) -> JobResponse: + """重试失败任务。 + + 将任务重置为 pending，retry_count + 1，但不自动 dispatch。 + 需要再次调用 /submit 提交执行。 + """ + use_case = RetryJobUseCase(job_repo) + + try: + job = use_case.execute(job_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id: + raise HTTPException(status_code=403, detail="Access denied to this job") + + return job_to_response(job) + + +@router.post("/jobs/{job_id}/cancel", response_model=JobResponse) +def cancel_job( + job_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + job_repo: Any = Depends(get_job_repository), +) -> JobResponse: + """取消任务。""" + use_case = CancelJobUseCase(job_repo) + + try: + job = use_case.execute(job_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id: + raise HTTPException(status_code=403, detail="Access denied to this job") + + return job_to_response(job) diff --git a/apps/api/app/dependencies.py b/apps/api/app/dependencies.py old mode 100644 new mode 100755 index 58f859960..82e81dadf --- a/apps/api/app/dependencies.py +++ b/apps/api/app/dependencies.py @@ -34,6 +34,7 @@ from packages.adapters.sqlalchemy_impl.generation_task_repository import ( from packages.adapters.sqlalchemy_impl.ingest_job_repository import ( SQLAlchemyIngestJobRepository, ) +from packages.adapters.sqlalchemy_impl.job_repository import SQLAlchemyJobRepository from packages.adapters.sqlalchemy_impl.project_repository import ( SQLAlchemyProjectRepository, ) @@ -52,6 +53,7 @@ from packages.ports.duplication_repository import DuplicationRecordRepository from packages.ports.generated_video_repository import GeneratedVideoRepository from packages.ports.generation_task_repository import GenerationTaskRepository from packages.ports.ingest_job_repository import IngestJobRepository +from packages.ports.job_repository import JobRepository from packages.ports.project_repository import ProjectRepository from packages.ports.title_library_repository import TitleLibraryRepository from packages.ports.user_repository import UserRepository @@ -104,6 +106,13 @@ def get_generation_task_repository( return SQLAlchemyGenerationTaskRepository(session) +def get_job_repository( + session: Session = Depends(get_db_session), +) -> SQLAlchemyJobRepository: + """Provide the SQLAlchemy job repository implementation.""" + return SQLAlchemyJobRepository(session) + + def get_generated_video_repository( session: Session = Depends(get_db_session), ) -> SQLAlchemyGeneratedVideoRepository: diff --git a/apps/api/app/schemas/job.py b/apps/api/app/schemas/job.py new file mode 100755 index 000000000..f678ddc89 --- /dev/null +++ b/apps/api/app/schemas/job.py @@ -0,0 +1,109 @@ +"""Job API schemas — Phase 8 任务 2.10.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Optional + +from pydantic import BaseModel, Field + + +class CreateJobRequest(BaseModel): + """创建任务请求体。""" + + project_id: str = Field(..., min_length=1, description="项目 ID") + job_type: str = Field( + ..., + description="任务类型: video_compose / render_edit_plan / asset_ingest / classification / voice_extraction / generation", + ) + payload: dict[str, Any] = Field(default_factory=dict, description="任务输入参数") + source_id: str = Field(default="", description="关联的业务实体 ID（如 edit_plan_id）") + max_retries: int = Field(default=3, ge=0, le=10, description="最大重试次数") + + +class UpdateProgressRequest(BaseModel): + """更新任务进度请求体。""" + + progress: float = Field(..., ge=0.0, le=100.0, description="进度百分比") + current_stage: str = Field(default="", description="当前阶段描述") + + +class CompleteJobRequest(BaseModel): + """完成任务请求体。""" + + result: dict[str, Any] = Field(default_factory=dict, description="任务结果") + + +class FailJobRequest(BaseModel): + """标记任务失败请求体。""" + + error_message: str = Field(..., min_length=1, description="错误信息") + + +class JobResponse(BaseModel): + """任务响应体。""" + + id: str + project_id: str + job_type: str + status: str + progress: float + current_stage: str + payload: dict[str, Any] + result: dict[str, Any] + error_message: str + retry_count: int + max_retries: int + celery_task_id: str + source_id: str + created_by_user_id: str + is_retryable: bool + started_at: Optional[datetime] = None + completed_at: Optional[datetime] = None + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} + + +class ListJobsResponse(BaseModel): + """任务列表响应体。""" + + items: list[JobResponse] + total: int + + +class JobStatisticsResponse(BaseModel): + """任务统计响应体。""" + + project_id: str + total: int + pending: int + running: int + success: int + failed: int + + +def job_to_response(job) -> JobResponse: + """将 Job 领域对象转换为 API 响应。""" + return JobResponse( + id=job.id, + project_id=job.project_id, + job_type=job.job_type.value if hasattr(job.job_type, "value") else str(job.job_type), + status=job.status.value if hasattr(job.status, "value") else str(job.status), + progress=job.progress, + current_stage=job.current_stage, + payload=job.payload, + result=job.result, + error_message=job.error_message, + retry_count=job.retry_count, + max_retries=job.max_retries, + celery_task_id=job.celery_task_id, + source_id=job.source_id, + created_by_user_id=job.created_by_user_id, + is_retryable=job.is_retryable, + started_at=job.started_at, + completed_at=job.completed_at, + created_at=job.created_at, + updated_at=job.updated_at, + ) diff --git a/apps/api/app/services/__init__.py b/apps/api/app/services/__init__.py old mode 100644 new mode 100755 index 8df5e4e17..b245a6023 --- a/apps/api/app/services/__init__.py +++ b/apps/api/app/services/__init__.py @@ -3,11 +3,13 @@ from .auto_clip_service import AutoClipService from .edit_plan_service import EditPlanService from .edit_template_service import EditTemplateService +from .job_service import JobService from .video_compose_service import VideoComposeService __all__ = [ "AutoClipService", "EditPlanService", "EditTemplateService", + "JobService", "VideoComposeService", ] diff --git a/apps/api/app/services/job_service.py b/apps/api/app/services/job_service.py new file mode 100755 index 000000000..f8800fd60 --- /dev/null +++ b/apps/api/app/services/job_service.py @@ -0,0 +1,268 @@ +"""JobService 服务层 — Phase 8 任务 2.10. + +将 JobService 与 VideoComposeService 集成，提供视频合成的完整异步工作流： +1. 创建 Job（记录任务元数据） +2. 提交执行（dispatch Celery 任务） +3. Celery 任务中更新进度、处理完成/失败 + +同时也提供通用的 Job 管理能力，供 ClipPlanService、RenderOrchestrator 等使用。 +""" + +from __future__ import annotations + +import logging +from typing import Any + +from sqlalchemy.orm import Session + +from packages.application.jobs import ( + CancelJobUseCase, + CompleteJobCommand, + CompleteJobUseCase, + CreateJobCommand, + CreateJobUseCase, + FailJobCommand, + FailJobUseCase, + GetJobStatisticsUseCase, + GetJobUseCase, + ListJobsUseCase, + RetryJobUseCase, + SubmitJobUseCase, + UpdateJobProgressCommand, + UpdateJobProgressUseCase, +) +from packages.domain.job import Job, JobStatus, JobType +from packages.ports.job_repository import JobRepository + +logger = logging.getLogger(__name__) + + +class JobService: + """统一异步任务管理服务。 + + 职责： + - 为视频合成等耗时操作提供统一的异步任务管理 + - 封装 Use Case 的调用，提供简洁的服务接口 + - 与 VideoComposeService 集成，支持视频合成工作流 + + 用法:: + + job_service = JobService(db) + job = job_service.create_compose_job( + project_id="xxx", + plan_id="yyy", + user_id="zzz", + ) + job_service.submit_job(job.id, celery_task_id="celery-xxx") + """ + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + # ── 创建任务 ────────────────────────────────────────────────────────── + + def create_compose_job( + self, + project_id: str, + plan_id: str, + user_id: str, + *, + max_retries: int = 3, + ) -> Job: + """创建视频合成任务。 + + Args: + project_id: 项目 ID + plan_id: EditPlan ID + user_id: 创建人 ID + max_retries: 最大重试次数 + + Returns: + 创建的 Job 实例 + """ + use_case = CreateJobUseCase(self._job_repo) + return use_case.execute( + CreateJobCommand( + project_id=project_id, + job_type=JobType.VIDEO_COMPOSE, + payload={"plan_id": plan_id}, + source_id=plan_id, + created_by_user_id=user_id, + max_retries=max_retries, + ) + ) + + def create_render_job( + self, + project_id: str, + plan_id: str, + user_id: str, + *, + max_retries: int = 3, + ) -> Job: + """创建剪辑计划渲染任务。""" + use_case = CreateJobUseCase(self._job_repo) + return use_case.execute( + CreateJobCommand( + project_id=project_id, + job_type=JobType.RENDER_EDIT_PLAN, + payload={"plan_id": plan_id}, + source_id=plan_id, + created_by_user_id=user_id, + max_retries=max_retries, + ) + ) + + def create_job( + self, + project_id: str, + job_type: JobType | str, + *, + payload: dict | None = None, + source_id: str = "", + user_id: str = "", + max_retries: int = 3, + ) -> Job: + """创建通用任务。""" + use_case = CreateJobUseCase(self._job_repo) + return use_case.execute( + CreateJobCommand( + project_id=project_id, + job_type=job_type, + payload=payload or {}, + source_id=source_id, + created_by_user_id=user_id, + max_retries=max_retries, + ) + ) + + # ── 提交执行 ────────────────────────────────────────────────────────── + + def submit_job(self, job_id: str, celery_task_id: str = "") -> Job: + """提交任务执行。""" + use_case = SubmitJobUseCase(self._job_repo) + return use_case.execute(job_id, celery_task_id) + + # ── 进度更新 ────────────────────────────────────────────────────────── + + def update_progress(self, job_id: str, progress: float, stage: str = "") -> Job: + """更新任务进度。""" + use_case = UpdateJobProgressUseCase(self._job_repo) + return use_case.execute( + UpdateJobProgressCommand( + job_id=job_id, + progress=progress, + current_stage=stage, + ) + ) + + # ── 完成 / 失败 ──────────────────────────────────────────────────────── + + def complete_job(self, job_id: str, result: dict | None = None) -> Job: + """标记任务完成。""" + use_case = CompleteJobUseCase(self._job_repo) + return use_case.execute(CompleteJobCommand(job_id=job_id, result=result or {})) + + def fail_job(self, job_id: str, error_message: str) -> Job: + """标记任务失败。""" + use_case = FailJobUseCase(self._job_repo) + return use_case.execute(FailJobCommand(job_id=job_id, error_message=error_message)) + + # ── 重试 / 取消 ──────────────────────────────────────────────────────── + + def retry_job(self, job_id: str) -> Job: + """重试失败任务。""" + use_case = RetryJobUseCase(self._job_repo) + return use_case.execute(job_id) + + def cancel_job(self, job_id: str) -> Job: + """取消任务。""" + use_case = CancelJobUseCase(self._job_repo) + return use_case.execute(job_id) + + # ── 查询 ───────────────────────────────────────────────────────────── + + def get_job(self, job_id: str) -> Job | None: + """获取任务详情。""" + use_case = GetJobUseCase(self._job_repo) + return use_case.execute(job_id) + + def list_project_jobs( + self, + project_id: str, + *, + job_type: JobType | str | None = None, + status: JobStatus | str | None = None, + limit: int = 50, + offset: int = 0, + ) -> list[Job]: + """获取项目下的任务列表。""" + use_case = ListJobsUseCase(self._job_repo) + return use_case.execute( + project_id=project_id, + job_type=job_type, + status=status, + limit=limit, + offset=offset, + ) + + def list_user_jobs( + self, + user_id: str, + *, + job_type: JobType | str | None = None, + status: JobStatus | str | None = None, + limit: int = 50, + offset: int = 0, + ) -> list[Job]: + """获取用户的任务列表。""" + use_case = ListJobsUseCase(self._job_repo) + return use_case.execute( + user_id=user_id, + job_type=job_type, + status=status, + limit=limit, + offset=offset, + ) + + def get_statistics(self, project_id: str) -> dict[str, Any]: + """获取项目任务统计。""" + use_case = GetJobStatisticsUseCase(self._job_repo) + return use_case.execute(project_id) + + # ── 防重复检查 ──────────────────────────────────────────────────────── + + def has_active_job_for_source(self, source_id: str, job_type: JobType | str) -> bool: + """检查是否已有活跃任务（防止重复提交）。 + + Args: + source_id: 关联的业务实体 ID + job_type: 任务类型 + + Returns: + True 如果存在活跃任务 + """ + return self._job_repo.find_active_by_source(source_id, job_type) is not None + + # ── 便捷方法：带防重的视频合成提交 ────────────────────────────────────── + + def submit_compose_if_not_exists( + self, + project_id: str, + plan_id: str, + user_id: str, + celery_task_id: str = "", + ) -> tuple[Job, bool]: + """创建并提交视频合成任务（防重复）。 + + Returns: + (job, created): job 实例和是否新创建的标志 + """ + if self.has_active_job_for_source(plan_id, JobType.VIDEO_COMPOSE): + existing = self._job_repo.find_active_by_source(plan_id, JobType.VIDEO_COMPOSE) + logger.info("已存在活跃的视频合成任务: job_id=%s plan_id=%s", existing.id, plan_id) + return existing, False + + job = self.create_compose_job(project_id, plan_id, user_id) + job = self.submit_job(job.id, celery_task_id) + return job, True diff --git a/apps/worker/worker_app/celery_app.py b/apps/worker/worker_app/celery_app.py old mode 100644 new mode 100755 index fabaec025..1c3f24870 --- a/apps/worker/worker_app/celery_app.py +++ b/apps/worker/worker_app/celery_app.py @@ -13,5 +13,6 @@ celery_app.conf.imports = ( "worker_app.tasks.generation", "worker_app.tasks.voice_extraction", "worker_app.tasks.edit_plan_generation", + "worker_app.tasks.compose_video", "apps.worker.video_processing.dedup", ) diff --git a/apps/worker/worker_app/tasks/__init__.py b/apps/worker/worker_app/tasks/__init__.py old mode 100644 new mode 100755 index a12198ea0..03e8cf67d --- a/apps/worker/worker_app/tasks/__init__.py +++ b/apps/worker/worker_app/tasks/__init__.py @@ -25,6 +25,10 @@ def __getattr__(name: str): from .voice_extraction import extract_voice_task return extract_voice_task + elif name == "compose_video": + from .compose_video import compose_video + + return compose_video elif name == "extract_background_task": from .voice_extraction import extract_background_task @@ -34,6 +38,7 @@ def __getattr__(name: str): __all__ = [ "classify_asset", + "compose_video", "generate_video", "healthcheck", "ingest_asset", diff --git a/apps/worker/worker_app/tasks/compose_video.py b/apps/worker/worker_app/tasks/compose_video.py new file mode 100755 index 000000000..d1e51ab9f --- /dev/null +++ b/apps/worker/worker_app/tasks/compose_video.py @@ -0,0 +1,136 @@ +"""视频合成 Celery 任务 — Phase 8 任务 2.10. + +使用 JobService 管理任务生命周期，集成 VideoComposeService 执行合成。 +""" + +from __future__ import annotations + +import logging +import shutil +import subprocess +import tempfile +from datetime import datetime, timezone +from pathlib import Path + +from celery.utils.log import get_task_logger +from worker_app.celery_app import celery_app +from worker_app.db import SessionLocal + +logger = get_task_logger(__name__) + + +def _get_job_service(): + """延迟导入 JobService，避免循环依赖。""" + from packages.adapters.sqlalchemy_impl.job_repository import SQLAlchemyJobRepository + from apps.api.app.services.job_service import JobService + + db = SessionLocal() + repo = SQLAlchemyJobRepository(db) + return JobService(repo), db + + +@celery_app.task( + name="worker.compose_video", + bind=True, + max_retries=3, + default_retry_delay=60, +) +def compose_video(self, job_id: str, **kwargs): + """视频合成任务。 + + Args: + job_id: JobService 中的任务 ID + **kwargs: 来自 Job.payload 的额外参数（plan_id, output_path 等） + """ + job_service, db = _get_job_service() + + try: + job = job_service.get_job(job_id) + if job is None: + logger.error("Job not found: %s", job_id) + return {"status": "error", "message": f"Job {job_id} not found"} + + plan_id = job.payload.get("plan_id", "") + if not plan_id: + job_service.fail_job(job_id, "Missing plan_id in job payload") + return {"status": "error", "message": "Missing plan_id"} + + # 标记为 running + job_service.update_progress(job_id, progress=10.0, current_stage="初始化合成环境") + + # 延迟导入 VideoComposeService + from apps.api.app.services.video_compose_service import VideoComposeService + + compose_svc = VideoComposeService(db) + + # 校验合成条件 + job_service.update_progress(job_id, progress=20.0, current_stage="校验合成条件") + validation = compose_svc.validate_compose(plan_id) + if not validation.valid: + error_msg = "; ".join(validation.errors) + job_service.fail_job(job_id, f"合成校验失败: {error_msg}") + return {"status": "error", "message": error_msg} + + # 构建合成命令 + job_service.update_progress(job_id, progress=30.0, current_stage="构建 FFmpeg 命令") + output_path = f"/tmp/video_output/{job_id}.mp4" + compose_cmd = compose_svc.build_compose_command(plan_id, output_path) + + # 执行 FFmpeg + job_service.update_progress(job_id, progress=50.0, current_stage="正在执行视频合成") + logger.info("Executing FFmpeg for job %s, plan %s", job_id, plan_id) + + try: + subprocess.run( + compose_cmd.command, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=3600, + ) + except subprocess.CalledProcessError as e: + job_service.fail_job(job_id, f"FFmpeg 执行失败: {e.stderr[:500]}") + raise + + # 上传结果 + job_service.update_progress(job_id, progress=80.0, current_stage="上传合成结果") + storage_key = f"rendered/{plan_id}/{job_id}.mp4" + + from worker_app.tasks.edit_plan_generation import _upload_to_oss + + output_url = _upload_to_oss(Path(output_path), storage_key) + + # 更新 Job 状态为完成 + result_data = { + "plan_id": plan_id, + "output_path": output_path, + "storage_key": storage_key, + "output_url": output_url or "", + "estimated_duration": compose_cmd.estimated_duration, + "clip_count": len(compose_cmd.clip_chains), + } + job_service.complete_job(job_id, result=result_data) + + logger.info("视频合成完成: job_id=%s, plan_id=%s", job_id, plan_id) + return {"status": "completed", "job_id": job_id, "result": result_data} + + except self.retry_exc as exc: + logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc) + raise + except Exception as exc: + logger.exception("视频合成异常: job_id=%s", job_id) + try: + job_service.fail_job(job_id, str(exc)[:500]) + except Exception: + logger.exception("更新 Job 失败状态时出错") + raise self.retry(exc=exc, countdown=60) + finally: + db.close() + # 清理临时文件 + try: + output_path = f"/tmp/video_output/{job_id}.mp4" + if Path(output_path).exists(): + Path(output_path).unlink() + except Exception: + pass diff --git a/packages/adapters/sqlalchemy_impl/job_repository.py b/packages/adapters/sqlalchemy_impl/job_repository.py new file mode 100755 index 000000000..ee2276846 --- /dev/null +++ b/packages/adapters/sqlalchemy_impl/job_repository.py @@ -0,0 +1,169 @@ +"""JobRepository SQLAlchemy 适配器 — Phase 8 任务 2.10.""" + +from __future__ import annotations + +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.models import JobModel +from packages.domain.job import Job, JobStatus, JobType + + +def _to_domain(model: JobModel) -> Job: + """ORM 模型 → 领域实体。""" + return Job( + id=model.id, + project_id=model.project_id, + job_type=JobType(model.job_type), + status=JobStatus(model.status), + progress=float(model.progress or 0.0), + current_stage=model.current_stage or "", + payload=dict(model.payload or {}), + result=dict(model.result or {}), + error_message=model.error_message or "", + retry_count=int(model.retry_count or 0), + max_retries=int(model.max_retries or 3), + celery_task_id=model.celery_task_id or "", + source_id=model.source_id or "", + created_by_user_id=model.created_by_user_id or "", + started_at=model.started_at, + completed_at=model.completed_at, + created_at=model.created_at, + updated_at=model.updated_at, + ) + + +def _to_model(job: Job) -> JobModel: + """领域实体 → ORM 模型。""" + return JobModel( + id=job.id, + project_id=job.project_id, + job_type=job.job_type.value, + status=job.status.value, + progress=job.progress, + current_stage=job.current_stage, + payload=job.payload, + result=job.result, + error_message=job.error_message, + retry_count=job.retry_count, + max_retries=job.max_retries, + celery_task_id=job.celery_task_id, + source_id=job.source_id, + created_by_user_id=job.created_by_user_id, + started_at=job.started_at, + completed_at=job.completed_at, + created_at=job.created_at, + updated_at=job.updated_at, + ) + + +class SQLAlchemyJobRepository: + """JobRepository 的 SQLAlchemy 实现。""" + + def __init__(self, session: Session): + self.session = session + + def create(self, job: Job) -> Job: + model = _to_model(job) + self.session.add(model) + self.session.commit() + return job + + def get(self, job_id: str) -> Job | None: + model = self.session.query(JobModel).filter(JobModel.id == job_id).first() + if model is None: + return None + return _to_domain(model) + + def update(self, job: Job) -> Job: + model = self.session.query(JobModel).filter(JobModel.id == job.id).first() + if model is None: + raise ValueError(f"Job {job.id} not found") + model.project_id = job.project_id + model.job_type = job.job_type.value + model.status = job.status.value + model.progress = job.progress + model.current_stage = job.current_stage + model.payload = job.payload + model.result = job.result + model.error_message = job.error_message + model.retry_count = job.retry_count + model.max_retries = job.max_retries + model.celery_task_id = job.celery_task_id + model.source_id = job.source_id + model.created_by_user_id = job.created_by_user_id + model.started_at = job.started_at + model.completed_at = job.completed_at + model.updated_at = job.updated_at + self.session.commit() + return job + + def list_by_project( + self, + project_id: str, + *, + job_type: JobType | str | None = None, + status: JobStatus | str | None = None, + limit: int = 50, + offset: int = 0, + ) -> list[Job]: + query = self.session.query(JobModel).filter(JobModel.project_id == project_id) + if job_type is not None: + jt = job_type.value if isinstance(job_type, JobType) else job_type + query = query.filter(JobModel.job_type == jt) + if status is not None: + st = status.value if isinstance(status, JobStatus) else status + query = query.filter(JobModel.status == st) + models = query.order_by(JobModel.created_at.desc()).offset(offset).limit(limit).all() + return [_to_domain(m) for m in models] + + def list_by_user( + self, + user_id: str, + *, + job_type: JobType | str | None = None, + status: JobStatus | str | None = None, + limit: int = 50, + offset: int = 0, + ) -> list[Job]: + query = self.session.query(JobModel).filter( + JobModel.created_by_user_id == user_id + ) + if job_type is not None: + jt = job_type.value if isinstance(job_type, JobType) else job_type + query = query.filter(JobModel.job_type == jt) + if status is not None: + st = status.value if isinstance(status, JobStatus) else status + query = query.filter(JobModel.status == st) + models = query.order_by(JobModel.created_at.desc()).offset(offset).limit(limit).all() + return [_to_domain(m) for m in models] + + def count_by_project( + self, + project_id: str, + *, + status: JobStatus | str | None = None, + ) -> int: + query = self.session.query(JobModel).filter(JobModel.project_id == project_id) + if status is not None: + st = status.value if isinstance(status, JobStatus) else status + query = query.filter(JobModel.status == st) + return query.count() + + def find_active_by_source( + self, + source_id: str, + job_type: JobType | str, + ) -> Job | None: + jt = job_type.value if isinstance(job_type, JobType) else job_type + model = ( + self.session.query(JobModel) + .filter( + JobModel.source_id == source_id, + JobModel.job_type == jt, + JobModel.status.in_(["pending", "running"]), + ) + .first() + ) + if model is None: + return None + return _to_domain(model) diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 8559e5ce4..7519d0ea9 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -384,3 +384,27 @@ class TemplateCategoryModel(Base): user_id = Column(String(36), nullable=False, index=True) name = Column(String(100), nullable=False) created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + +class JobModel(Base): + """Phase 8 任务 2.10 — 统一异步任务 ORM 模型。""" + + __tablename__ = "jobs" + + id = Column(String(32), primary_key=True) + project_id = Column(String(32), nullable=False, index=True) + job_type = Column(String(30), nullable=False, index=True) + status = Column(String(20), nullable=False, default="pending", index=True) + progress = Column(Float, nullable=False, default=0.0) + current_stage = Column(String(200), nullable=False, default="") + payload = Column(JSON, nullable=False, default=dict) + result = Column(JSON, nullable=False, default=dict) + error_message = Column(Text, nullable=False, default="") + retry_count = Column(Integer, nullable=False, default=0) + max_retries = Column(Integer, nullable=False, default=3) + celery_task_id = Column(String(100), nullable=False, default="") + source_id = Column(String(32), nullable=False, default="", index=True) + created_by_user_id = Column(String(32), nullable=False, default="", index=True) + started_at = Column(DateTime, nullable=True) + completed_at = Column(DateTime, nullable=True) + created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) \ No newline at end of file diff --git a/packages/application/__init__.py b/packages/application/__init__.py old mode 100644 new mode 100755 index faabb13fc..978b6e345 --- a/packages/application/__init__.py +++ b/packages/application/__init__.py @@ -30,33 +30,63 @@ from .generation_tasks import ( GetGenerationTaskUseCase, ) from .ingest_jobs import SubmitIngestJobCommand, SubmitIngestJobUseCase +from .jobs import ( + CancelJobUseCase, + CompleteJobCommand, + CompleteJobUseCase, + CreateJobCommand, + CreateJobUseCase, + FailJobCommand, + FailJobUseCase, + GetJobStatisticsUseCase, + GetJobUseCase, + ListJobsUseCase, + RetryJobUseCase, + SubmitJobUseCase, + UpdateJobProgressCommand, + UpdateJobProgressUseCase, +) from .projects import CreateProjectCommand, CreateProjectUseCase, GetProjectUseCase, ListProjectsUseCase __all__ = [ + "CancelJobUseCase", + "CompleteJobCommand", + "CompleteJobUseCase", "CreateAssetCommand", "CreateAssetLibraryCommand", "CreateAssetLibraryUseCase", "CreateAssetUseCase", "CreateGenerationTaskCommand", "CreateGenerationTaskUseCase", + "CreateJobCommand", + "CreateJobUseCase", "CreateProjectCommand", "CreateProjectUseCase", "DeleteDuplicationRecordUseCase", + "FailJobCommand", + "FailJobUseCase", "GetDuplicationDetailUseCase", "GetGeneratedVideoDownloadUrlUseCase", "GetGeneratedVideoUseCase", + "GetJobStatisticsUseCase", + "GetJobUseCase", "GetProjectUseCase", "ListAssetLibrariesUseCase", "ListAssetsUseCase", "ListDuplicationRecordsUseCase", "ListGeneratedVideosByTaskUseCase", "ListGeneratedVideosUseCase", + "ListJobsUseCase", "ListProjectsUseCase", "RetryDuplicationUseCase", + "RetryJobUseCase", "SubmitClassificationJobCommand", "SubmitClassificationJobUseCase", "SubmitIngestJobCommand", "SubmitIngestJobUseCase", + "SubmitJobUseCase", + "UpdateJobProgressCommand", + "UpdateJobProgressUseCase", "UploadForDuplicationCommand", "UploadForDuplicationUseCase", ] diff --git a/packages/application/jobs.py b/packages/application/jobs.py new file mode 100755 index 000000000..ffec662f3 --- /dev/null +++ b/packages/application/jobs.py @@ -0,0 +1,257 @@ +"""JobService 应用层用例 — Phase 8 任务 2.10. + +提供任务创建、提交、状态管理、进度追踪、失败重试、查询等核心能力。 +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any +from uuid import uuid4 + +from packages.domain.job import Job, JobStatus, JobType +from packages.ports.job_repository import JobRepository + +logger = logging.getLogger(__name__) + + +# ── 命令对象 ────────────────────────────────────────────────────────────────── + + +@dataclass(slots=True) +class CreateJobCommand: + """创建任务的命令。""" + + project_id: str + job_type: JobType | str + payload: dict = field(default_factory=dict) + source_id: str = "" + created_by_user_id: str = "" + max_retries: int = 3 + + +@dataclass(slots=True) +class UpdateJobProgressCommand: + """更新任务进度的命令。""" + + job_id: str + progress: float + current_stage: str = "" + + +@dataclass(slots=True) +class CompleteJobCommand: + """完成任务的命令。""" + + job_id: str + result: dict = field(default_factory=dict) + + +@dataclass(slots=True) +class FailJobCommand: + """标记任务失败的命令。""" + + job_id: str + error_message: str + + +# ── 用例 ───────────────────────────────────────────────────────────────────── + + +class CreateJobUseCase: + """创建异步任务。 + + 创建后状态为 pending，不自动提交到 Celery（由调用方决定何时 dispatch）。 + """ + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute(self, command: CreateJobCommand) -> Job: + job = Job.create( + project_id=command.project_id, + job_type=command.job_type, + payload=command.payload, + source_id=command.source_id, + created_by_user_id=command.created_by_user_id, + max_retries=command.max_retries, + ) + logger.info( + "创建任务: job_id=%s type=%s project=%s", + job.id, job.job_type.value, job.project_id, + ) + return self._job_repo.create(job) + + +class SubmitJobUseCase: + """提交任务执行：将任务状态切换为 running 并返回 Celery 任务 ID 占位。 + + 实际的 Celery dispatch 由上层（路由层 / 服务层）完成，本用例仅做状态流转。 + """ + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute(self, job_id: str, celery_task_id: str = "") -> Job: + job = self._job_repo.get(job_id) + if job is None: + raise ValueError(f"任务不存在: {job_id}") + if job.status != JobStatus.PENDING: + raise ValueError(f"只有 pending 状态的任务可以提交，当前状态: {job.status.value}") + + job.mark_running(stage="已提交，等待执行") + job.celery_task_id = celery_task_id + return self._job_repo.update(job) + + +class UpdateJobProgressUseCase: + """更新任务进度。""" + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute(self, command: UpdateJobProgressCommand) -> Job: + job = self._job_repo.get(command.job_id) + if job is None: + raise ValueError(f"任务不存在: {command.job_id}") + if job.status != JobStatus.RUNNING: + raise ValueError(f"只有 running 状态的任务可以更新进度，当前状态: {job.status.value}") + + job.update_progress(command.progress, command.current_stage) + return self._job_repo.update(job) + + +class CompleteJobUseCase: + """标记任务成功完成。""" + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute(self, command: CompleteJobCommand) -> Job: + job = self._job_repo.get(command.job_id) + if job is None: + raise ValueError(f"任务不存在: {command.job_id}") + if job.status not in (JobStatus.RUNNING, JobStatus.PENDING): + raise ValueError(f"只有 running/pending 状态的任务可以标记完成，当前状态: {job.status.value}") + + job.mark_success(result=command.result) + logger.info("任务完成: job_id=%s", job.id) + return self._job_repo.update(job) + + +class FailJobUseCase: + """标记任务失败。""" + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute(self, command: FailJobCommand) -> Job: + job = self._job_repo.get(command.job_id) + if job is None: + raise ValueError(f"任务不存在: {command.job_id}") + + job.mark_failed(command.error_message) + logger.warning("任务失败: job_id=%s error=%s", job.id, command.error_message) + return self._job_repo.update(job) + + +class RetryJobUseCase: + """重试失败任务。 + + 将 failed 任务重置为 pending，retry_count + 1。 + 实际的 Celery 重新 dispatch 由调用方完成。 + """ + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute(self, job_id: str) -> Job: + job = self._job_repo.get(job_id) + if job is None: + raise ValueError(f"任务不存在: {job_id}") + + job.prepare_retry() + logger.info("任务重试: job_id=%s retry_count=%d", job.id, job.retry_count) + return self._job_repo.update(job) + + +class CancelJobUseCase: + """取消任务。""" + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute(self, job_id: str) -> Job: + job = self._job_repo.get(job_id) + if job is None: + raise ValueError(f"任务不存在: {job_id}") + if job.is_terminal: + raise ValueError(f"任务已处于终态，无法取消: {job.status.value}") + + job.mark_cancelled() + logger.info("任务取消: job_id=%s", job.id) + return self._job_repo.update(job) + + +class GetJobUseCase: + """获取单个任务详情。""" + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute(self, job_id: str) -> Job | None: + return self._job_repo.get(job_id) + + +class ListJobsUseCase: + """列出任务（支持按项目 / 用户 / 类型 / 状态过滤）。""" + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute( + self, + *, + project_id: str | None = None, + user_id: str | None = None, + job_type: JobType | str | None = None, + status: JobStatus | str | None = None, + limit: int = 50, + offset: int = 0, + ) -> list[Job]: + if project_id: + return self._job_repo.list_by_project( + project_id, job_type=job_type, status=status, + limit=limit, offset=offset, + ) + if user_id: + return self._job_repo.list_by_user( + user_id, job_type=job_type, status=status, + limit=limit, offset=offset, + ) + raise ValueError("必须指定 project_id 或 user_id") + + +class GetJobStatisticsUseCase: + """获取项目的任务统计摘要。""" + + def __init__(self, job_repo: JobRepository): + self._job_repo = job_repo + + def execute(self, project_id: str) -> dict[str, Any]: + total = self._job_repo.count_by_project(project_id) + pending = self._job_repo.count_by_project(project_id, status=JobStatus.PENDING) + running = self._job_repo.count_by_project(project_id, status=JobStatus.RUNNING) + success = self._job_repo.count_by_project(project_id, status=JobStatus.SUCCESS) + failed = self._job_repo.count_by_project(project_id, status=JobStatus.FAILED) + + return { + "project_id": project_id, + "total": total, + "pending": pending, + "running": running, + "success": success, + "failed": failed, + } diff --git a/packages/domain/__init__.py b/packages/domain/__init__.py old mode 100644 new mode 100755 index b9fb81f2a..e0ce5822d --- a/packages/domain/__init__.py +++ b/packages/domain/__init__.py @@ -23,6 +23,7 @@ from .entities import ( ) from .generated_video import GeneratedVideo from .generation_task import GenerationTask, GenerationTaskStatus +from .job import Job, JobStatus, JobType from .template_clip_config import ClipType, TemplateClipConfig, TransitionEffect from .title_library import TitleLibraryItem from .voice_library import VoiceLibraryItem @@ -51,6 +52,9 @@ __all__ = [ "GenerationTaskStatus", "IngestJob", "IngestJobStatus", + "Job", + "JobStatus", + "JobType", "Project", "TemplateClipConfig", "TransitionEffect", diff --git a/packages/domain/job.py b/packages/domain/job.py new file mode 100755 index 000000000..85a2fd7f5 --- /dev/null +++ b/packages/domain/job.py @@ -0,0 +1,289 @@ +"""Job 领域模型 — Phase 8 任务 2.10. + +统一的异步任务抽象，为视频合成等耗时操作提供异步任务管理能力。 + +状态机： + pending → running → success + ↘ failed → pending (重试) + ↘ cancelled +""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass, field +from datetime import datetime, timezone + +if sys.version_info >= (3, 11): + from enum import StrEnum +else: + from enum import Enum + + class StrEnum(str, Enum): + pass + + +from uuid import uuid4 + + +class JobType(StrEnum): + """任务类型枚举。""" + + VIDEO_COMPOSE = "video_compose" + """视频合成（VideoComposeService）""" + + RENDER_EDIT_PLAN = "render_edit_plan" + """剪辑计划渲染""" + + ASSET_INGEST = "asset_ingest" + """素材导入""" + + CLASSIFICATION = "classification" + """素材分类""" + + VOICE_EXTRACTION = "voice_extraction" + """语音提取""" + + GENERATION = "generation" + """通用生成""" + + +class JobStatus(StrEnum): + """任务状态枚举。""" + + PENDING = "pending" + RUNNING = "running" + SUCCESS = "success" + FAILED = "failed" + CANCELLED = "cancelled" + + +# 终态集合 +TERMINAL_STATUSES = frozenset({JobStatus.SUCCESS, JobStatus.FAILED, JobStatus.CANCELLED}) + +# 合法状态转换 +_VALID_TRANSITIONS: dict[JobStatus, set[JobStatus]] = { + JobStatus.PENDING: {JobStatus.RUNNING, JobStatus.SUCCESS, JobStatus.CANCELLED}, + JobStatus.RUNNING: {JobStatus.SUCCESS, JobStatus.FAILED, JobStatus.CANCELLED}, + JobStatus.FAILED: {JobStatus.PENDING}, # 重试回到 pending +} + + +@dataclass(slots=True) +class Job: + """统一异步任务实体。 + + Attributes: + id: 任务唯一标识 + project_id: 所属项目 + job_type: 任务类型 + status: 当前状态 + progress: 进度百分比 (0.0 ~ 100.0) + current_stage: 当前阶段描述（人类可读） + payload: 任务输入参数（JSON 序列化） + result: 任务结果（JSON 序列化） + error_message: 错误信息 + retry_count: 已重试次数 + max_retries: 最大重试次数 + celery_task_id: Celery 异步任务 ID + source_id: 关联的业务实体 ID（如 edit_plan_id, generation_task_id） + created_by_user_id: 创建人 + started_at: 开始执行时间 + completed_at: 完成时间 + created_at: 创建时间 + updated_at: 最后更新时间 + """ + + id: str + project_id: str + job_type: JobType + status: JobStatus = JobStatus.PENDING + progress: float = 0.0 + current_stage: str = "" + payload: dict = field(default_factory=dict) + result: dict = field(default_factory=dict) + error_message: str = "" + retry_count: int = 0 + max_retries: int = 3 + celery_task_id: str = "" + source_id: str = "" + created_by_user_id: str = "" + started_at: datetime | None = None + completed_at: datetime | None = None + created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + + @classmethod + def create( + cls, + project_id: str, + job_type: JobType | str, + *, + payload: dict | None = None, + source_id: str = "", + created_by_user_id: str = "", + max_retries: int = 3, + ) -> Job: + """创建新任务。 + + Args: + project_id: 项目 ID + job_type: 任务类型 + payload: 任务输入参数 + source_id: 关联业务实体 ID + created_by_user_id: 创建人 ID + max_retries: 最大重试次数 + + Returns: + 新建的 Job 实例 + + Raises: + ValueError: 参数校验失败 + """ + if not project_id.strip(): + raise ValueError("project_id 不能为空") + + # 兼容字符串类型 + if isinstance(job_type, str): + try: + job_type = JobType(job_type) + except ValueError: + raise ValueError(f"不支持的任务类型: {job_type}") + + return cls( + id=uuid4().hex, + project_id=project_id.strip(), + job_type=job_type, + payload=payload or {}, + source_id=source_id.strip(), + created_by_user_id=created_by_user_id.strip(), + max_retries=max_retries, + ) + + @property + def is_terminal(self) -> bool: + """是否处于终态。""" + return self.status in TERMINAL_STATUSES + + @property + def is_retryable(self) -> bool: + """是否可重试（失败且未超过重试上限）。""" + return self.status == JobStatus.FAILED and self.retry_count < self.max_retries + + def transition_to(self, new_status: JobStatus | str) -> None: + """执行状态转换。 + + Args: + new_status: 目标状态 + + Raises: + ValueError: 非法状态转换 + """ + if isinstance(new_status, str): + try: + new_status = JobStatus(new_status) + except ValueError: + raise ValueError(f"无效状态: {new_status}") + + allowed = _VALID_TRANSITIONS.get(self.status, set()) + if new_status not in allowed: + raise ValueError( + f"非法状态转换: {self.status.value} → {new_status.value}，" + f"允许: {{{', '.join(s.value for s in allowed)}}}" + ) + + now = datetime.now(timezone.utc) + self.status = new_status + self.updated_at = now + + # 自动设置时间戳 + if new_status == JobStatus.RUNNING and self.started_at is None: + self.started_at = now + elif new_status in (JobStatus.SUCCESS, JobStatus.FAILED): + self.completed_at = now + + def mark_running(self, stage: str = "") -> None: + """标记为执行中。""" + self.transition_to(JobStatus.RUNNING) + if stage: + self.current_stage = stage + + def mark_success(self, result: dict | None = None) -> None: + """标记为成功。""" + self.transition_to(JobStatus.SUCCESS) + self.progress = 100.0 + self.current_stage = "完成" + if result: + self.result = result + + def mark_failed(self, error_message: str) -> None: + """标记为失败。""" + self.transition_to(JobStatus.FAILED) + self.error_message = error_message + self.current_stage = "失败" + + def mark_cancelled(self) -> None: + """标记为取消。""" + self.transition_to(JobStatus.CANCELLED) + self.current_stage = "已取消" + + def update_progress(self, progress: float, stage: str = "") -> None: + """更新进度。 + + Args: + progress: 进度百分比 (0.0 ~ 100.0) + stage: 当前阶段描述 + + Raises: + ValueError: 进度值非法 + """ + if not 0.0 <= progress <= 100.0: + raise ValueError(f"进度必须在 0~100 之间，当前: {progress}") + self.progress = progress + if stage: + self.current_stage = stage + self.updated_at = datetime.now(timezone.utc) + + def prepare_retry(self) -> None: + """准备重试：重置状态为 pending。 + + Raises: + ValueError: 不可重试 + """ + if not self.is_retryable: + raise ValueError( + f"任务不可重试: status={self.status.value}, " + f"retry_count={self.retry_count}, max_retries={self.max_retries}" + ) + self.retry_count += 1 + self.transition_to(JobStatus.PENDING) + self.progress = 0.0 + self.current_stage = f"第 {self.retry_count} 次重试" + self.error_message = "" + self.started_at = None + self.completed_at = None + self.celery_task_id = "" + + def to_dict(self) -> dict: + """序列化为字典。""" + return { + "id": self.id, + "project_id": self.project_id, + "job_type": self.job_type.value, + "status": self.status.value, + "progress": self.progress, + "current_stage": self.current_stage, + "payload": self.payload, + "result": self.result, + "error_message": self.error_message, + "retry_count": self.retry_count, + "max_retries": self.max_retries, + "celery_task_id": self.celery_task_id, + "source_id": self.source_id, + "created_by_user_id": self.created_by_user_id, + "is_retryable": self.is_retryable, + "started_at": self.started_at.isoformat() if self.started_at else None, + "completed_at": self.completed_at.isoformat() if self.completed_at else None, + "created_at": self.created_at.isoformat() if self.created_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + } diff --git a/packages/ports/job_repository.py b/packages/ports/job_repository.py new file mode 100755 index 000000000..59119bb3c --- /dev/null +++ b/packages/ports/job_repository.py @@ -0,0 +1,64 @@ +"""JobRepository 端口接口 — Phase 8 任务 2.10.""" + +from __future__ import annotations + +from typing import Protocol + +from packages.domain.job import Job, JobStatus, JobType + + +class JobRepository(Protocol): + """异步任务仓储接口。""" + + def create(self, job: Job) -> Job: + """持久化一个新任务。""" + ... + + def get(self, job_id: str) -> Job | None: + """根据 ID 获取任务。""" + ... + + def update(self, job: Job) -> Job: + """更新任务状态。""" + ... + + def list_by_project( + self, + project_id: str, + *, + job_type: JobType | str | None = None, + status: JobStatus | str | None = None, + limit: int = 50, + offset: int = 0, + ) -> list[Job]: + """按项目列出任务，支持类型和状态过滤。""" + ... + + def list_by_user( + self, + user_id: str, + *, + job_type: JobType | str | None = None, + status: JobStatus | str | None = None, + limit: int = 50, + offset: int = 0, + ) -> list[Job]: + """按创建人列出任务。""" + ... + + def count_by_project( + self, + project_id: str, + *, + status: JobStatus | str | None = None, + ) -> int: + """按项目统计任务数量。""" + ... + + def find_active_by_source( + self, + source_id: str, + job_type: JobType | str, + ) -> Job | None: + """查找关联同一业务实体的活跃任务（pending/running）。""" + ... diff --git a/tests/unit/test_job_service.py b/tests/unit/test_job_service.py new file mode 100755 index 000000000..aceb69c3e --- /dev/null +++ b/tests/unit/test_job_service.py @@ -0,0 +1,578 @@ +"""Unit tests for JobService — Phase 8 任务 2.10. + +测试 Job 领域模型、Use Cases 和 Service 层。 +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path + +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") +os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) + +from packages.domain.job import Job, JobStatus, JobType +from packages.application.jobs import ( + CancelJobUseCase, + CompleteJobCommand, + CompleteJobUseCase, + CreateJobCommand, + CreateJobUseCase, + FailJobCommand, + FailJobUseCase, + GetJobStatisticsUseCase, + GetJobUseCase, + ListJobsUseCase, + RetryJobUseCase, + SubmitJobUseCase, + UpdateJobProgressCommand, + UpdateJobProgressUseCase, +) + + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +class FakeJobRepo: + """内存中的 JobRepository 测试替身。""" + + def __init__(self): + self._store: dict[str, Job] = {} + + def create(self, job: Job) -> Job: + self._store[job.id] = job + return job + + def get(self, job_id: str) -> Job | None: + return self._store.get(job_id) + + def update(self, job: Job) -> Job: + if job.id not in self._store: + raise ValueError(f"Job {job.id} not found") + self._store[job.id] = job + return job + + def list_by_project(self, project_id, *, job_type=None, status=None, limit=50, offset=0): + results = [ + j for j in self._store.values() + if j.project_id == project_id + and (job_type is None or j.job_type == job_type or j.job_type == JobType(job_type)) + and (status is None or j.status == status or j.status == JobStatus(status)) + ] + return results[offset : offset + limit] + + def list_by_user(self, user_id, *, job_type=None, status=None, limit=50, offset=0): + results = [ + j for j in self._store.values() + if j.created_by_user_id == user_id + and (job_type is None or j.job_type == job_type or j.job_type == JobType(job_type)) + and (status is None or j.status == status or j.status == JobStatus(status)) + ] + return results[offset : offset + limit] + + def count_by_project(self, project_id, *, status=None): + return len([ + j for j in self._store.values() + if j.project_id == project_id + and (status is None or j.status == status or j.status == JobStatus(status)) + ]) + + def find_active_by_source(self, source_id, job_type): + jt = job_type.value if isinstance(job_type, JobType) else job_type + for j in self._store.values(): + if j.source_id == source_id and j.job_type.value == jt and j.status in (JobStatus.PENDING, JobStatus.RUNNING): + return j + return None + + +@pytest.fixture +def repo(): + return FakeJobRepo() + + +# ── Job 领域模型测试 ────────────────────────────────────────────────────────── + + +class TestJobDomainModel: + def test_create_job(self): + job = Job.create( + project_id="proj-1", + job_type=JobType.VIDEO_COMPOSE, + payload={"plan_id": "plan-1"}, + source_id="plan-1", + created_by_user_id="user-1", + ) + assert job.project_id == "proj-1" + assert job.job_type == JobType.VIDEO_COMPOSE + assert job.status == JobStatus.PENDING + assert job.progress == 0.0 + assert job.retry_count == 0 + assert job.max_retries == 3 + assert job.payload == {"plan_id": "plan-1"} + + def test_create_job_with_string_type(self): + job = Job.create(project_id="proj-1", job_type="render_edit_plan") + assert job.job_type == JobType.RENDER_EDIT_PLAN + + def test_create_job_invalid_type_raises(self): + with pytest.raises(ValueError, match="不支持的任务类型"): + Job.create(project_id="proj-1", job_type="invalid_type") + + def test_create_job_empty_project_id_raises(self): + with pytest.raises(ValueError, match="project_id 不能为空"): + Job.create(project_id=" ", job_type=JobType.VIDEO_COMPOSE) + + def test_is_terminal(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + assert not job.is_terminal + + job.status = JobStatus.SUCCESS + assert job.is_terminal + + job.status = JobStatus.FAILED + assert job.is_terminal + + job.status = JobStatus.CANCELLED + assert job.is_terminal + + def test_is_retryable(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=2) + job.status = JobStatus.FAILED + assert job.is_retryable + + job.retry_count = 2 + assert not job.is_retryable + + def test_mark_running(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running("初始化中") + assert job.status == JobStatus.RUNNING + assert job.current_stage == "初始化中" + assert job.started_at is not None + + def test_mark_success(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_success(result={"output_url": "https://example.com/video.mp4"}) + assert job.status == JobStatus.SUCCESS + assert job.progress == 100.0 + assert job.current_stage == "完成" + assert job.result == {"output_url": "https://example.com/video.mp4"} + assert job.completed_at is not None + + def test_mark_failed(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_failed("FFmpeg 执行失败") + assert job.status == JobStatus.FAILED + assert job.error_message == "FFmpeg 执行失败" + assert job.completed_at is not None + + def test_mark_cancelled(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + job.mark_cancelled() + assert job.status == JobStatus.CANCELLED + assert job.current_stage == "已取消" + + def test_update_progress(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + job.update_progress(50.0, "合成中") + assert job.progress == 50.0 + assert job.current_stage == "合成中" + + def test_update_progress_invalid_value_raises(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + with pytest.raises(ValueError, match="进度必须在 0~100 之间"): + job.update_progress(150.0) + + def test_prepare_retry(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=3) + job.mark_running() + job.mark_failed("临时错误") + job.prepare_retry() + assert job.status == JobStatus.PENDING + assert job.retry_count == 1 + assert job.progress == 0.0 + assert job.error_message == "" + assert "第 1 次重试" in job.current_stage + + def test_prepare_retry_exceeded_raises(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=1) + job.mark_running() + job.mark_failed("错误") + job.prepare_retry() # retry_count = 1, max_retries = 1 → not retryable + with pytest.raises(ValueError, match="任务不可重试"): + job.prepare_retry() + + def test_invalid_transition_raises(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_success() + with pytest.raises(ValueError, match="非法状态转换"): + job.transition_to(JobStatus.RUNNING) # success → running 不合法（终态不可转换） + + def test_to_dict(self): + job = Job.create(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + d = job.to_dict() + assert d["project_id"] == "proj-1" + assert d["job_type"] == "video_compose" + assert d["status"] == "pending" + assert "is_retryable" in d + + +# ── Use Case 测试 ───────────────────────────────────────────────────────────── + + +class TestCreateJobUseCase: + def test_execute(self, repo): + use_case = CreateJobUseCase(repo) + cmd = CreateJobCommand( + project_id="proj-1", + job_type=JobType.VIDEO_COMPOSE, + payload={"plan_id": "plan-1"}, + source_id="plan-1", + created_by_user_id="user-1", + ) + job = use_case.execute(cmd) + assert job.id in repo._store + assert job.status == JobStatus.PENDING + + def test_execute_with_string_type(self, repo): + use_case = CreateJobUseCase(repo) + cmd = CreateJobCommand( + project_id="proj-1", + job_type="video_compose", + ) + job = use_case.execute(cmd) + assert job.job_type == JobType.VIDEO_COMPOSE + + +class TestSubmitJobUseCase: + def test_submit_pending_job(self, repo): + # 先创建 + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + + # 提交 + submit_uc = SubmitJobUseCase(repo) + submitted = submit_uc.execute(job.id, celery_task_id="celery-abc") + assert submitted.status == JobStatus.RUNNING + assert submitted.celery_task_id == "celery-abc" + assert submitted.started_at is not None + + def test_submit_non_pending_raises(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + job.mark_running() + repo.update(job) + + submit_uc = SubmitJobUseCase(repo) + with pytest.raises(ValueError, match="只有 pending 状态"): + submit_uc.execute(job.id) + + def test_submit_nonexistent_raises(self, repo): + submit_uc = SubmitJobUseCase(repo) + with pytest.raises(ValueError, match="任务不存在"): + submit_uc.execute("nonexistent-id") + + +class TestUpdateJobProgressUseCase: + def test_update_running_job(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + job.mark_running() + repo.update(job) + + progress_uc = UpdateJobProgressUseCase(repo) + updated = progress_uc.execute( + UpdateJobProgressCommand(job_id=job.id, progress=75.0, current_stage="渲染中") + ) + assert updated.progress == 75.0 + assert updated.current_stage == "渲染中" + + def test_update_non_running_raises(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + + progress_uc = UpdateJobProgressUseCase(repo) + with pytest.raises(ValueError, match="只有 running 状态"): + progress_uc.execute( + UpdateJobProgressCommand(job_id=job.id, progress=50.0) + ) + + +class TestCompleteJobUseCase: + def test_complete_running_job(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + job.mark_running() + repo.update(job) + + complete_uc = CompleteJobUseCase(repo) + completed = complete_uc.execute( + CompleteJobCommand(job_id=job.id, result={"url": "https://example.com/v.mp4"}) + ) + assert completed.status == JobStatus.SUCCESS + assert completed.progress == 100.0 + assert completed.result == {"url": "https://example.com/v.mp4"} + + def test_complete_non_running_raises(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + job.mark_running() + job.mark_success() # 已经是终态 + + complete_uc = CompleteJobUseCase(repo) + with pytest.raises(ValueError, match="只有 running/pending 状态"): + complete_uc.execute(CompleteJobCommand(job_id=job.id)) + + +class TestFailJobUseCase: + def test_fail_job(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + job.mark_running() + repo.update(job) + + fail_uc = FailJobUseCase(repo) + failed = fail_uc.execute(FailJobCommand(job_id=job.id, error_message="磁盘空间不足")) + assert failed.status == JobStatus.FAILED + assert failed.error_message == "磁盘空间不足" + + +class TestRetryJobUseCase: + def test_retry_failed_job(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, max_retries=3)) + job.mark_running() + job.mark_failed("网络超时") + repo.update(job) + + retry_uc = RetryJobUseCase(repo) + retried = retry_uc.execute(job.id) + assert retried.status == JobStatus.PENDING + assert retried.retry_count == 1 + assert retried.error_message == "" + + def test_retry_non_failed_raises(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + + retry_uc = RetryJobUseCase(repo) + with pytest.raises(ValueError, match="任务不可重试"): + retry_uc.execute(job.id) + + +class TestCancelJobUseCase: + def test_cancel_pending_job(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + + cancel_uc = CancelJobUseCase(repo) + cancelled = cancel_uc.execute(job.id) + assert cancelled.status == JobStatus.CANCELLED + + def test_cancel_completed_raises(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + job.mark_running() + job.mark_success() + repo.update(job) + + cancel_uc = CancelJobUseCase(repo) + with pytest.raises(ValueError, match="已处于终态"): + cancel_uc.execute(job.id) + + +class TestGetJobUseCase: + def test_get_existing_job(self, repo): + create_uc = CreateJobUseCase(repo) + job = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + + get_uc = GetJobUseCase(repo) + fetched = get_uc.execute(job.id) + assert fetched is not None + assert fetched.id == job.id + + def test_get_nonexistent_returns_none(self, repo): + get_uc = GetJobUseCase(repo) + assert get_uc.execute("nonexistent") is None + + +class TestListJobsUseCase: + def test_list_by_project(self, repo): + create_uc = CreateJobUseCase(repo) + create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.RENDER_EDIT_PLAN)) + create_uc.execute(CreateJobCommand(project_id="proj-2", job_type=JobType.VIDEO_COMPOSE)) + + list_uc = ListJobsUseCase(repo) + jobs = list_uc.execute(project_id="proj-1") + assert len(jobs) == 2 + + def test_list_by_project_with_type_filter(self, repo): + create_uc = CreateJobUseCase(repo) + create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.RENDER_EDIT_PLAN)) + + list_uc = ListJobsUseCase(repo) + jobs = list_uc.execute(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE) + assert len(jobs) == 1 + assert jobs[0].job_type == JobType.VIDEO_COMPOSE + + def test_list_by_project_with_status_filter(self, repo): + create_uc = CreateJobUseCase(repo) + j1 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + j2 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + j2.mark_running() + repo.update(j2) + + list_uc = ListJobsUseCase(repo) + running_jobs = list_uc.execute(project_id="proj-1", status=JobStatus.RUNNING) + assert len(running_jobs) == 1 + + def test_list_by_user(self, repo): + create_uc = CreateJobUseCase(repo) + create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, created_by_user_id="user-1")) + create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE, created_by_user_id="user-2")) + + list_uc = ListJobsUseCase(repo) + jobs = list_uc.execute(user_id="user-1") + assert len(jobs) == 1 + + def test_list_without_project_or_user_raises(self, repo): + list_uc = ListJobsUseCase(repo) + with pytest.raises(ValueError, match="必须指定"): + list_uc.execute() + + +class TestGetJobStatisticsUseCase: + def test_statistics(self, repo): + create_uc = CreateJobUseCase(repo) + j1 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + j2 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + j2.mark_running() + repo.update(j2) + j3 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + j3.mark_running() + j3.mark_success() + repo.update(j3) + j4 = create_uc.execute(CreateJobCommand(project_id="proj-1", job_type=JobType.VIDEO_COMPOSE)) + j4.mark_running() + j4.mark_failed("err") + repo.update(j4) + + stats_uc = GetJobStatisticsUseCase(repo) + stats = stats_uc.execute("proj-1") + assert stats["total"] == 4 + assert stats["pending"] == 1 + assert stats["running"] == 1 + assert stats["success"] == 1 + assert stats["failed"] == 1 + + +# ── JobService 集成测试 ─────────────────────────────────────────────────────── + + +class TestJobService: + def test_create_compose_job(self, repo): + from app.services.job_service import JobService + + svc = JobService(repo) + job = svc.create_compose_job( + project_id="proj-1", + plan_id="plan-1", + user_id="user-1", + ) + assert job.job_type == JobType.VIDEO_COMPOSE + assert job.payload == {"plan_id": "plan-1"} + assert job.source_id == "plan-1" + assert job.created_by_user_id == "user-1" + + def test_create_render_job(self, repo): + from app.services.job_service import JobService + + svc = JobService(repo) + job = svc.create_render_job( + project_id="proj-1", + plan_id="plan-1", + user_id="user-1", + ) + assert job.job_type == JobType.RENDER_EDIT_PLAN + + def test_submit_compose_if_not_exists_creates_new(self, repo): + from app.services.job_service import JobService + + svc = JobService(repo) + job, created = svc.submit_compose_if_not_exists("proj-1", "plan-1", "user-1", "celery-1") + assert created is True + assert job.status == JobStatus.RUNNING + assert job.celery_task_id == "celery-1" + + def test_submit_compose_if_not_exists_returns_existing(self, repo): + from app.services.job_service import JobService + + svc = JobService(repo) + job1, created1 = svc.submit_compose_if_not_exists("proj-1", "plan-1", "user-1", "celery-1") + job2, created2 = svc.submit_compose_if_not_exists("proj-1", "plan-1", "user-1", "celery-2") + assert created2 is False + assert job2.id == job1.id + + def test_full_lifecycle(self, repo): + """完整生命周期测试：创建 → 提交 → 更新进度 → 完成。""" + from app.services.job_service import JobService + + svc = JobService(repo) + + # 创建 + job = svc.create_compose_job("proj-1", "plan-1", "user-1") + assert job.status == JobStatus.PENDING + + # 提交 + job = svc.submit_job(job.id, "celery-xyz") + assert job.status == JobStatus.RUNNING + assert job.celery_task_id == "celery-xyz" + + # 更新进度 + job = svc.update_progress(job.id, 50.0, "合成中") + assert job.progress == 50.0 + + job = svc.update_progress(job.id, 80.0, "上传结果") + assert job.progress == 80.0 + + # 完成 + job = svc.complete_job(job.id, result={"url": "https://example.com/v.mp4"}) + assert job.status == JobStatus.SUCCESS + assert job.progress == 100.0 + + def test_full_lifecycle_with_retry(self, repo): + """带重试的完整生命周期测试。""" + from app.services.job_service import JobService + + svc = JobService(repo) + + # 创建并执行 + job = svc.create_compose_job("proj-1", "plan-1", "user-1", max_retries=2) + job = svc.submit_job(job.id, "celery-1") + + # 失败 + job = svc.fail_job(job.id, "临时网络错误") + assert job.status == JobStatus.FAILED + + # 重试 + job = svc.retry_job(job.id) + assert job.status == JobStatus.PENDING + assert job.retry_count == 1 + + # 再次提交 + job = svc.submit_job(job.id, "celery-2") + assert job.status == JobStatus.RUNNING + + # 成功 + job = svc.complete_job(job.id, result={"url": "ok"}) + assert job.status == JobStatus.SUCCESS -- 2.54.0 From 89a53b52111cfa2d947ddbd42005bbcefb32bc7f Mon Sep 17 00:00:00 2001 From: CI Test Date: Wed, 1 Jul 2026 23:45:07 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20submit/retry/ca?= =?UTF-8?q?ncel=20=E8=B7=AF=E7=94=B1=E6=9D=83=E9=99=90=E6=A3=80=E6=9F=A5?= =?UTF-8?q?=E9=A1=BA=E5=BA=8F=EF=BC=88P1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将权限检查从状态变更之后移到之前，防止非授权用户触发状态变更： - submit_job: 先获取 job 并验证权限，再执行 SubmitJobUseCase - retry_job: 同上 - cancel_job: 同上 修复审计 P1 问题。 --- apps/api/app/api/routes/jobs.py | 31 +++++++++++++++++++++---------- 1 file changed, 21 insertions(+), 10 deletions(-) diff --git a/apps/api/app/api/routes/jobs.py b/apps/api/app/api/routes/jobs.py index ee8554ab4..84b72163f 100755 --- a/apps/api/app/api/routes/jobs.py +++ b/apps/api/app/api/routes/jobs.py @@ -130,6 +130,13 @@ def submit_job( 将任务状态从 pending 切换为 running，并 dispatch Celery 异步任务。 """ + # 权限检查：先获取任务并验证权限，再执行状态变更 + job = job_repo.get(job_id) + if job is None: + raise HTTPException(status_code=404, detail=f"Job {job_id} not found") + if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id: + raise HTTPException(status_code=403, detail="Access denied to this job") + use_case = SubmitJobUseCase(job_repo) try: @@ -137,10 +144,6 @@ def submit_job( except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) - # 权限检查 - if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id: - raise HTTPException(status_code=403, detail="Access denied to this job") - # Dispatch Celery 任务 celery_task_name = _JOB_TYPE_TO_CELERY_TASK.get(job.job_type.value) if celery_task_name: @@ -290,6 +293,13 @@ def retry_job( 将任务重置为 pending，retry_count + 1，但不自动 dispatch。 需要再次调用 /submit 提交执行。 """ + # 权限检查：先获取任务并验证权限，再执行状态变更 + job = job_repo.get(job_id) + if job is None: + raise HTTPException(status_code=404, detail=f"Job {job_id} not found") + if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id: + raise HTTPException(status_code=403, detail="Access denied to this job") + use_case = RetryJobUseCase(job_repo) try: @@ -297,9 +307,6 @@ def retry_job( except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) - if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id: - raise HTTPException(status_code=403, detail="Access denied to this job") - return job_to_response(job) @@ -310,6 +317,13 @@ def cancel_job( job_repo: Any = Depends(get_job_repository), ) -> JobResponse: """取消任务。""" + # 权限检查：先获取任务并验证权限，再执行状态变更 + job = job_repo.get(job_id) + if job is None: + raise HTTPException(status_code=404, detail=f"Job {job_id} not found") + if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id: + raise HTTPException(status_code=403, detail="Access denied to this job") + use_case = CancelJobUseCase(job_repo) try: @@ -317,7 +331,4 @@ def cancel_job( except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) - if job.created_by_user_id and job.created_by_user_id != authenticated_user.user.id: - raise HTTPException(status_code=403, detail="Access denied to this job") - return job_to_response(job) -- 2.54.0