# Export Service 技术栈合规性重构

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

---

## 变更概述

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

---

## 主要变更

### 1. 日志系统 ✅

**之前**:
```python
# 没有日志记录
```

**现在**:
```python
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. 异步数据库操作 ✅

**之前**:
```python
def __init__(self, db: Session):
    self.db = db
```

**现在**:
```python
def __init__(
    self,
    async_session: AsyncSession,
    export_job_repository: ExportJobRepository,
    project_repository: ProjectRepository
):
    self.async_session = async_session
```

### 3. 依赖注入模式 ✅

**之前**:
```python
# Service 内部创建 Repository
self.job_repository = ExportJobRepository(db)
```

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

### 4. 错误处理 ✅

**之前**:
```python
from app.core.exceptions import NotFoundError, PermissionError
```

**现在**:
```python
from app.core.exceptions import (
    ResourceNotFoundError,
    PermissionDeniedError,
    ValidationError
)
```

### 5. API 响应格式 ✅

**之前**:
```python
return {
    'job_id': job.id,
    'task_id': task.id,
    'status': 'pending'
}
```

**现在**:
```python
return SuccessResponse(data=ExportCreateResponse(
    job_id=job.id,
    task_id=task.id,
    status=ExportStatusEnum.PENDING
))
```

### 6. Pydantic Schema 定义 ✅

**新增**:
```python
# 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. 数据模型 ✅

**新增**:
```python
# 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 任务 ✅

**之前**:
```python
from app.tasks.celery_app import celery_app
db = SessionLocal()  # 同步数据库
```

**现在**:
```python
from app.core.celery_app import celery_app
from app.core.database import AsyncSessionLocal

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

### 9. API 路由实现 ✅

**新增完整路由**:
```python
# 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. 时间戳处理 ✅

**之前**:
```python
'created_at': job.created_at.isoformat()
```

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

### 11. 类型注解 ✅

**完整类型注解**:
```python
async def create_export_job(
    self,
    user_id: str,
    request: ExportCreateRequest
) -> ExportCreateResponse:
```

### 12. FFmpeg 错误处理 ✅

**新增**:
```python
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**:
```python
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. 数据库迁移 ✅

**新增迁移脚本模板**:
```python
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 |
| 类型注解 | ✅ | 完整的类型提示 |

---

## 文档结构优化

### 新增章节
1. **数据模型**: 完整的 SQLModel 定义
2. **Pydantic Schemas**: 请求/响应 Schema
3. **Repository 层**: 数据访问层实现
4. **Service 层**: 业务逻辑层实现
5. **API 路由**: FastAPI 路由实现
6. **Celery 任务**: 异步任务实现
7. **视频合成工具**: FFmpeg 集成
8. **错误处理**: 错误码和日志
9. **数据库迁移**: Alembic 迁移脚本

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

---

## 影响范围

### 需要创建的文件
1. `app/models/export_job.py` - 数据模型
2. `app/schemas/export.py` - Pydantic Schemas
3. `app/repositories/export_job_repository.py` - Repository
4. `app/services/export_service.py` - Service
5. `app/api/v1/export.py` - API 路由
6. `app/tasks/export_tasks.py` - Celery 任务
7. `app/utils/video_composer.py` - 视频合成工具
8. `alembic/versions/xxx_create_export_jobs_table.py` - 数据库迁移

### 需要更新的文件
1. `app/api/v1/__init__.py` - 注册 export 路由
2. `server/requirements.txt` - 添加 ffmpeg-python 依赖

---

## 后续工作

### 1. 实现代码
- [ ] 创建所有模型、Schema、Repository、Service 文件
- [ ] 实现 API 路由
- [ ] 实现 Celery 任务
- [ ] 实现视频合成工具

### 2. 数据库迁移
- [ ] 创建 export_jobs 表迁移脚本
- [ ] 执行迁移

### 3. 测试
- [ ] 单元测试（Repository、Service）
- [ ] 集成测试（API）
- [ ] Celery 任务测试

### 4. 依赖安装
```bash
pip install ffmpeg-python
```

### 5. 系统依赖
```bash
# 安装 FFmpeg
apt-get install ffmpeg  # Ubuntu/Debian
brew install ffmpeg     # macOS
```

---

## 相关文档

- [jointo-tech-stack Skill](/.claude/skills/jointo-tech-stack/SKILL.md)
- [Backend 规范](/.claude/skills/jointo-tech-stack/references/backend.md)
- [Database 规范](/.claude/skills/jointo-tech-stack/references/database.md)
- [API 设计规范](/.claude/skills/jointo-tech-stack/references/api-design.md)
- [日志系统迁移](../rfcs/201-logging-system-migration.md)

---

**变更作者**: Kiro AI  
**审核状态**: 待审核  
**优先级**: 高