8.2 KiB

Raw Permalink Blame History

Export Service 技术栈合规性重构

日期: 2026-01-29
类型: 文档重构
影响范围: docs/requirements/backend/04-services/project/export-service.md

变更概述

完全重写 Export Service 文档，使其符合 jointo-tech-stack 规范，修复了 14 个不合规问题。

主要变更

1. 日志系统 ✅

之前:

# 没有日志记录

现在:

from app.core.logging import get_logger

logger = get_logger(__name__)

logger.info(
    "export_job_created",
    job_id=job.id,
    project_id=project_id,
    user_id=user_id
)

2. 异步数据库操作 ✅

之前:

def __init__(self, db: Session):
    self.db = db

现在:

def __init__(
    self,
    async_session: AsyncSession,
    export_job_repository: ExportJobRepository,
    project_repository: ProjectRepository
):
    self.async_session = async_session

3. 依赖注入模式 ✅

之前:

# Service 内部创建 Repository
self.job_repository = ExportJobRepository(db)

现在:

# 通过构造函数注入
def __init__(
    self,
    async_session: AsyncSession,
    export_job_repository: ExportJobRepository,
    project_repository: ProjectRepository
):

4. 错误处理 ✅

之前:

from app.core.exceptions import NotFoundError, PermissionError

现在:

from app.core.exceptions import (
    ResourceNotFoundError,
    PermissionDeniedError,
    ValidationError
)

5. API 响应格式 ✅

之前:

return {
    'job_id': job.id,
    'task_id': task.id,
    'status': 'pending'
}

现在:

return SuccessResponse(data=ExportCreateResponse(
    job_id=job.id,
    task_id=task.id,
    status=ExportStatusEnum.PENDING
))

6. Pydantic Schema 定义 ✅

新增:

# app/schemas/export.py
class ExportCreateRequest(BaseModel):
    project_id: str
    format: ExportFormatEnum
    quality: ExportQualityEnum

class ExportJobResponse(BaseModel):
    id: str
    project_id: str
    status: ExportStatusEnum
    # ... 完整类型定义

7. 数据模型 ✅

新增:

# app/models/export_job.py
class ExportStatus(IntEnum):
    PENDING = 1
    PROCESSING = 2
    COMPLETED = 3
    FAILED = 4

class ExportJob(SQLModel, table=True):
    __tablename__ = "export_jobs"
    __table_args__ = {"comment": "视频导出任务表"}
    
    id: str = Field(primary_key=True)
    status: int = Field(sa_column=Column(SmallInteger))
    # ... 使用 TIMESTAMP(timezone=True)

8. Celery 任务 ✅

之前:

from app.tasks.celery_app import celery_app
db = SessionLocal()  # 同步数据库

现在:

from app.core.celery_app import celery_app
from app.core.database import AsyncSessionLocal

async with AsyncSessionLocal() as session:
    # 异步数据库操作

9. API 路由实现 ✅

新增完整路由:

# app/api/v1/export.py
@router.post("", response_model=SuccessResponse[ExportCreateResponse])
async def create_export_job(
    request: ExportCreateRequest,
    current_user: UserResponse = Depends(get_current_user),
    export_service: ExportService = Depends(get_export_service)
):
    result = await export_service.create_export_job(
        user_id=current_user.id,
        request=request
    )
    return SuccessResponse(data=result)

10. 时间戳处理 ✅

之前:

'created_at': job.created_at.isoformat()

现在:

# Pydantic 自动序列化
created_at: datetime = Field(..., description="创建时间")

11. 类型注解 ✅

完整类型注解:

async def create_export_job(
    self,
    user_id: str,
    request: ExportCreateRequest
) -> ExportCreateResponse:

12. FFmpeg 错误处理 ✅

新增:

try:
    ffmpeg.run(output, capture_stdout=True, capture_stderr=True)
except ffmpeg.Error as e:
    logger.error(
        "ffmpeg_error",
        stderr=e.stderr.decode('utf-8'),
        exc_info=True
    )
    raise ValueError(f"视频合成失败: {e.stderr.decode('utf-8')}")

13. 枚举类型处理 ✅

使用 SMALLINT + IntEnum:

class ExportStatus(IntEnum):
    PENDING = 1
    PROCESSING = 2
    COMPLETED = 3
    FAILED = 4

# 数据库字段
status: int = Field(
    sa_column=Column(SmallInteger, nullable=False),
    description="任务状态: 1=pending, 2=processing, 3=completed, 4=failed"
)

14. 数据库迁移 ✅

新增迁移脚本模板:

def upgrade() -> None:
    op.create_table(
        'export_jobs',
        sa.Column('id', UUID(as_uuid=False), primary_key=True),
        sa.Column('status', sa.SmallInteger(), nullable=False),
        sa.Column('created_at', TIMESTAMP(timezone=True), ...),
        comment='视频导出任务表'
    )

新增内容

1. Repository 层完整实现

ExportJobRepository 类
异步 CRUD 操作
结构化日志记录

2. Service 层完整实现

ExportService 类
依赖注入
权限检查
枚举转换方法

3. API 路由完整实现

创建导出任务
查询任务详情
获取任务列表
取消任务

4. Celery 任务完整实现

export_video_task
export_json_task
异步数据库操作
错误处理和重试

5. 视频合成工具完整实现

VideoComposer 类
FFmpeg 集成
错误处理
结构化日志

6. 数据模型定义

ExportJob 模型
枚举类型定义
数据库迁移脚本

技术栈合规性检查

检查项	状态	说明
异步数据库	✅	使用 AsyncSession + asyncpg
日志系统	✅	使用 structlog + get_logger
依赖注入	✅	Repository 通过构造函数注入
错误处理	✅	使用标准异常类
API 响应	✅	使用 SuccessResponse 包装
Schema 定义	✅	完整的 Pydantic Schema
枚举类型	✅	SMALLINT + IntEnum
时间戳	✅	TIMESTAMP(timezone=True)
UUID	✅	UUID v7
Celery	✅	使用 app.core.celery_app
类型注解	✅	完整的类型提示

文档结构优化

新增章节

数据模型: 完整的 SQLModel 定义
Pydantic Schemas: 请求/响应 Schema
Repository 层: 数据访问层实现
Service 层: 业务逻辑层实现
API 路由: FastAPI 路由实现
Celery 任务: 异步任务实现
视频合成工具: FFmpeg 集成
错误处理: 错误码和日志
数据库迁移: Alembic 迁移脚本

文档元信息

版本号: v1.0 → v2.0
更新日期: 2025-01-27 → 2026-01-29
新增合规状态标识

影响范围

需要创建的文件

app/models/export_job.py - 数据模型
app/schemas/export.py - Pydantic Schemas
app/repositories/export_job_repository.py - Repository
app/services/export_service.py - Service
app/api/v1/export.py - API 路由
app/tasks/export_tasks.py - Celery 任务
app/utils/video_composer.py - 视频合成工具
alembic/versions/xxx_create_export_jobs_table.py - 数据库迁移

需要更新的文件

app/api/v1/__init__.py - 注册 export 路由
server/requirements.txt - 添加 ffmpeg-python 依赖

后续工作

1. 实现代码

创建所有模型、Schema、Repository、Service 文件
实现 API 路由
实现 Celery 任务
实现视频合成工具

2. 数据库迁移

创建 export_jobs 表迁移脚本
执行迁移

3. 测试

单元测试（Repository、Service）
集成测试（API）
Celery 任务测试

4. 依赖安装

pip install ffmpeg-python

5. 系统依赖

# 安装 FFmpeg
apt-get install ffmpeg  # Ubuntu/Debian
brew install ffmpeg     # macOS

8.2 KiB Raw Permalink Blame History