# 技术栈选型

> **文档版本**：v1.0  
> **最后更新**：2025-01-27

---

## 目录

1. [核心框架](#核心框架)
2. [异步任务处理](#异步任务处理)
3. [缓存系统](#缓存系统)
4. [文件存储](#文件存储)
5. [视频处理](#视频处理)
6. [AI 服务集成](#ai-服务集成)
7. [其他工具](#其他工具)

---

## 核心框架

### Web 框架：FastAPI

**选择理由**：

- ✅ 高性能（基于 Starlette 和 Pydantic）
- ✅ 自动生成 API 文档（Swagger/OpenAPI）
- ✅ 类型提示支持（与 TypeScript 前端对接友好）
- ✅ 异步支持（async/await）
- ✅ 现代化设计，易于维护

**版本**：FastAPI 0.104+

**示例代码**：

```python
from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session

app = FastAPI(
    title="Jointo API",
    version="1.0.0",
    docs_url="/api/docs",
    redoc_url="/api/redoc"
)

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
```

---

### ORM：SQLModel

**选择理由**：

- ✅ **FastAPI 作者开发**，与 FastAPI 完美集成
- ✅ **类型安全**：基于 Pydantic v2，编译时类型检查
- ✅ **减少重复代码**：单一模型定义，同时用于 ORM 和 API Schema
- ✅ **底层稳定**：基于 SQLAlchemy 2.0，继承其强大功能
- ✅ **异步支持**：原生 async/await
- ✅ **数据库迁移**：兼容 Alembic

**版本**：SQLModel 0.0.14+

**核心优势**：

```python
# 传统 SQLAlchemy 需要定义两次
# models/project.py
class ProjectModel(Base):
    __tablename__ = "projects"
    id = Column(String, primary_key=True)
    name = Column(String)

# schemas/project.py
class ProjectSchema(BaseModel):
    id: str
    name: str

# SQLModel 只需定义一次 ✅
from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List

class Project(SQLModel, table=True):
    id: Optional[str] = Field(default=None, primary_key=True)
    name: str = Field(index=True, max_length=255)
    description: Optional[str] = None
    type: str  # 'mine' | 'collab'
    created_at: Optional[datetime] = Field(default_factory=datetime.utcnow)
    updated_at: Optional[datetime] = Field(default_factory=datetime.utcnow)

    # 关联关系
    storyboards: List["Storyboard"] = Relationship(back_populates="project")

# 直接用于 API，无需额外 Schema
@app.post("/projects", response_model=Project)
async def create_project(project: Project):
    # project 既是数据库模型，也是 Pydantic 模型
    return project
```

**与 SQLAlchemy 的关系**：

- SQLModel 是 SQLAlchemy 的高级封装
- 底层使用 SQLAlchemy 2.0 引擎
- 可以混用 SQLAlchemy 的高级特性
- 完全兼容 Alembic 迁移工具

---

### 数据库迁移：Alembic

**选择理由**：

- ✅ SQLAlchemy/SQLModel 官方迁移工具
- ✅ 版本控制数据库结构
- ✅ 自动生成迁移脚本
- ✅ 完全兼容 SQLModel

**版本**：Alembic 1.12+

**与 SQLModel 集成**：

```python
# alembic/env.py
from sqlmodel import SQLModel
from app.models import Project, Storyboard  # 导入所有模型

target_metadata = SQLModel.metadata  # 使用 SQLModel 的 metadata
```

**常用命令**：

```bash
# 初始化 Alembic
alembic init alembic

# 自动生成迁移脚本
alembic revision --autogenerate -m "Add users table"

# 执行迁移
alembic upgrade head

# 回滚迁移
alembic downgrade -1
```

---

## 异步任务处理

### 任务队列：Celery

**选择理由**：

- ✅ Python 生态最成熟的任务队列
- ✅ 支持多种消息代理（RabbitMQ、Redis）
- ✅ 支持任务重试、优先级、定时任务
- ✅ 丰富的监控工具（Flower）

**版本**：Celery 5.3+

**示例代码**：

```python
from celery import Celery

celery_app = Celery(
    "jointo",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0"
)

@celery_app.task(bind=True, max_retries=3)
def generate_video_task(self, prompt: str, video_type: str):
    try:
        # AI 生成视频逻辑
        result = ai_service.generate_video(prompt, video_type)
        return result
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)
```

---

### 消息代理：RabbitMQ（推荐）或 Redis

**RabbitMQ**：

- ✅ 功能强大，支持复杂路由
- ✅ 适合生产环境
- ✅ 支持消息持久化

**Redis**：

- ✅ 简单轻量
- ✅ 适合小规模部署
- ✅ 可同时用作缓存

**推荐**：生产环境使用 RabbitMQ，开发环境使用 Redis

---

## 缓存系统

### Redis

**用途**：

- ✅ 缓存热点数据（项目列表、分镜列表）
- ✅ 会话存储
- ✅ Celery 消息代理（可选）
- ✅ 分布式锁
- ✅ 实时数据（在线用户数等）

**版本**：Redis 7.0+

**示例代码**：

```python
import redis
from typing import Optional

class CacheService:
    def __init__(self):
        self.client = redis.Redis(
            host='localhost',
            port=6379,
            db=0,
            decode_responses=True
        )

    def get(self, key: str) -> Optional[str]:
        return self.client.get(key)

    def set(self, key: str, value: str, ttl: int = 300):
        self.client.setex(key, ttl, value)

    def delete(self, key: str):
        self.client.delete(key)
```

---

## 文件存储

### 对象存储：MinIO（开发）或 AWS S3 / 阿里云 OSS（生产）

**选择理由**：

- ✅ 支持 S3 兼容 API
- ✅ 适合存储大量文件（图片、视频、音频）
- ✅ 支持 CDN 加速
- ✅ 成本低，扩展性好

**MinIO**：用于开发环境，本地部署  
**AWS S3 / 阿里云 OSS**：用于生产环境

**示例代码**：

```python
from minio import Minio

client = Minio(
    "localhost:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
    secure=False
)

# 上传文件
client.fput_object(
    "jointo",
    "videos/sample.mp4",
    "/tmp/sample.mp4",
    content_type="video/mp4"
)
```

---

## 视频处理

### FFmpeg

**用途**：

- ✅ 视频合成
- ✅ 视频转码
- ✅ 视频裁剪
- ✅ 音频处理

**Python 封装**：`ffmpeg-python`

**示例代码**：

```python
import ffmpeg

# 合成视频
(
    ffmpeg
    .input('input1.mp4')
    .concat(
        ffmpeg.input('input2.mp4')
    )
    .output('output.mp4')
    .run()
)

# 添加音频
(
    ffmpeg
    .input('video.mp4')
    .input('audio.mp3')
    .output('output.mp4', vcodec='copy', acodec='aac')
    .run()
)
```

---

## AI 服务集成

### 图片生成

**选项**：

- **Stable Diffusion**（开源）：自部署，成本低
- **DALL-E API**（OpenAI）：质量高，按次计费
- **Midjourney API**（如果可用）：艺术风格强

**推荐**：Stable Diffusion（开发）+ DALL-E（生产）

---

### 视频生成

**选项**：

- **Runway Gen-2**：商业服务，质量高
- **Pika Labs**：新兴服务，效果好
- **Stable Video Diffusion**：开源方案

**推荐**：Runway Gen-2（主要）+ Stable Video Diffusion（备选）

---

### 语音合成

**选项**：

- **OpenAI TTS**：质量高，多语言支持
- **Azure Speech Services**：企业级，稳定性好
- **百度语音 / 讯飞语音**：中文优化，价格低

**推荐**：OpenAI TTS（主要）+ 百度语音（备选）

---

### 字幕生成

**选项**：

- **Whisper**（OpenAI，开源）：准确率高，支持多语言
- **阿里云语音识别**：中文优化

**推荐**：Whisper（自部署）

---

## 其他工具

### 认证授权：JWT

**库**：`python-jose[cryptography]`

**示例代码**：

```python
from jose import jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=30)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
```

---

### 环境配置：Pydantic Settings

**库**：`pydantic-settings`

**示例代码**：

```python
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    redis_url: str
    secret_key: str

    class Config:
        env_file = ".env"

settings = Settings()
```

---

### 日志：Loguru

**库**：`loguru`

**示例代码**：

```python
from loguru import logger

logger.add("logs/app.log", rotation="500 MB")
logger.info("Application started")
logger.error("An error occurred")
```

---

### 监控：Prometheus + Grafana（可选）

**用途**：

- 监控系统性能
- 追踪 API 响应时间
- 监控任务队列状态

---

## 技术栈总结

| 层级     | 技术选型       | 版本    |
| -------- | -------------- | ------- |
| 编程语言 | Python         | 3.12+   |
| Web 框架 | FastAPI        | 0.104+  |
| ORM      | SQLModel       | 0.0.14+ |
| 数据库   | PostgreSQL     | 17      |
| 缓存     | Redis          | 7+      |
| 任务队列 | Celery         | 5.3+    |
| 消息代理 | RabbitMQ       | 3.12+   |
| 文件存储 | MinIO / AWS S3 | -       |
| 视频处理 | FFmpeg         | 5.0+    |
| 认证     | JWT            | -       |
| 容器化   | Docker         | 24+     |

---

## 相关文档

- [架构概述](./01-architecture-overview.md)
- [系统架构设计](./03-system-design.md)
- [异步任务处理](./07-async-tasks.md)

---

**文档版本**：v1.0  
**最后更新**：2025-01-27