feat(app): add FastAPI main application and configuration

- Create main.py with CORS and GZip middleware
- Add Settings class with all configuration options
- Support .env file for environment variables
- Add health check and root endpoints
- Create comprehensive README with quick start guide
- Add .env.example template
- Include API usage examples and troubleshooting

Phase 4 Task 30/68 completed
This commit is contained in:
Xiaoxia AI
2026-06-17 07:42:22 +08:00
parent 267f097a7b
commit 37836b2755
4 changed files with 329 additions and 451 deletions
+30 -83
View File
@@ -1,85 +1,32 @@
# =======================
# 应用基础配置
# =======================
APP_ENV=development # development / staging / production
APP_NAME=xiaoxia-saas
APP_VERSION=0.1.0
# 小虾 SaaS 环境变量配置
# ==================== 应用配置 ====================
APP_NAME=小虾 SaaS
BASE_URL=http://localhost:3000
# ==================== 数据库配置 ====================
DATABASE_URL=postgresql://xiaoxia_user:your_password@localhost:5432/xiaoxia_saas
# ==================== Redis 配置 ====================
REDIS_URL=redis://localhost:6379/0
# ==================== JWT 配置 ====================
JWT_SECRET_KEY=your-super-secret-key-change-this-in-production-min-32-chars
JWT_ALGORITHM=HS256
JWT_ACCESS_TOKEN_EXPIRE_MINUTES=30
JWT_REFRESH_TOKEN_EXPIRE_DAYS=30
# ==================== 邮件配置 ====================
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your-email@gmail.com
SMTP_PASSWORD=your-app-specific-password
SMTP_FROM_EMAIL=noreply@xiaoxia-saas.com
SMTP_FROM_NAME=小虾 SaaS
# ==================== 环境配置 ====================
ENVIRONMENT=development
DEBUG=true
# =======================
# API 服务配置
# =======================
API_HOST=0.0.0.0
API_PORT=8000
API_PREFIX=/api/v1
# =======================
# Web 前端配置
# =======================
WEB_PORT=3000
WEB_URL=http://localhost:3000
# =======================
# 数据库配置
# =======================
DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/xiaoxia_saas
DATABASE_POOL_SIZE=20
DATABASE_MAX_OVERFLOW=40
DATABASE_POOL_TIMEOUT=30
DATABASE_POOL_RECYCLE=3600
# =======================
# Redis 配置
# =======================
REDIS_URL=redis://localhost:6379/0
REDIS_MAX_CONNECTIONS=50
# =======================
# Celery Worker 配置
# =======================
CELERY_BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/1
CELERY_WORKER_CONCURRENCY=4
CELERY_WORKER_MAX_TASKS_PER_CHILD=1000
# =======================
# MinIO 对象存储配置
# =======================
MINIO_ENDPOINT=localhost:9000
MINIO_ACCESS_KEY=CHANGE_ME
MINIO_SECRET_KEY=CHANGE_ME
MINIO_BUCKET=xiaoxia-assets
MINIO_SECURE=false
MINIO_PUBLIC_URL=http://localhost:9000
# =======================
# 日志配置
# =======================
LOG_LEVEL=INFO # DEBUG / INFO / WARNING / ERROR / CRITICAL
LOG_FORMAT=json # json / text
LOG_FILE=/var/log/xiaoxia-saas/app.log
# =======================
# CORS 配置
# =======================
CORS_ORIGINS=http://localhost:3000,http://localhost:8000
CORS_ALLOW_CREDENTIALS=true
# =======================
# 文件上传限制
# =======================
MAX_UPLOAD_SIZE_MB=1000
ALLOWED_FILE_TYPES=video/mp4,video/quicktime,video/x-msvideo,audio/mpeg,audio/wav,image/jpeg,image/png,image/gif
# =======================
# 安全配置
# =======================
SECRET_KEY=change-me-in-production-to-a-random-string-at-least-32-chars
ACCESS_TOKEN_EXPIRE_MINUTES=60
REFRESH_TOKEN_EXPIRE_DAYS=7
# =======================
# 监控与追踪(可选)
# =======================
# SENTRY_DSN=
# PROMETHEUS_PORT=9090
# ==================== CORS 配置 ====================
CORS_ORIGINS=["http://localhost:3000","http://localhost:5173"]
+189 -308
View File
@@ -1,332 +1,213 @@
# Xiaoxia SaaS - AI 视频自动化剪辑系统
# 小虾 SaaS - 快速开始指南
新一代 SaaS 版小虾自动化剪辑系统,采用 Clean Architecture 重新设计与实现。
## 🚀 快速启动
## 项目状态
**当前阶段**: 核心业务骨架已完成(Phase 1)
- ✅ Clean Architecture 架构就绪
- ✅ 核心业务对象(User, Workspace, Project, AssetLibrary, Asset, IngestJob, ClassificationJob)
- ✅ 完整上传→入库→Asset 创建链路
- ✅ 完整分类任务链路
- ✅ 双持久化实现(In-Memory + PostgreSQL)
- ✅ Alembic 数据库迁移
- ✅ 8 个集成测试全绿
- ✅ Docker Compose 开发环境
## 技术栈
**Backend**
- Python 3.12
- FastAPI - REST API
- Pydantic - 数据验证
- SQLAlchemy - ORM
- Alembic - 数据库迁移
**Worker**
- Celery - 异步任务队列
- Redis - 消息队列
**Database**
- PostgreSQL - 生产数据库
- SQLite - 测试环境
**Frontend** (占位)
- Next.js 14
- TypeScript
- React 18
**Architecture**
- Clean Architecture
- Ports & Adapters (Hexagonal)
- Repository Pattern
- Use Case Pattern
**Testing**
- pytest
- 集成测试优先策略
## 项目结构
```
xiaoxia-saas/
├── packages/ # 共享业务逻辑包
│ ├── domain/ # 核心业务实体与规则
│ ├── application/ # 用例层
│ ├── ports/ # 接口定义
│ └── adapters/ # 接口实现
│ ├── in_memory/ # 内存实现(测试用)
│ └── sqlalchemy_impl/ # PostgreSQL 实现
├── apps/ # 应用层
│ ├── api/ # FastAPI REST API
│ ├── worker/ # Celery 异步任务
│ └── web/ # Next.js 前端(占位)
├── infra/ # 基础设施配置
│ ├── docker/ # Docker 配置
│ └── nginx/ # Nginx 配置
├── tests/ # 测试
│ ├── integration/ # 集成测试
│ └── e2e/ # 端到端测试(占位)
├── alembic/ # 数据库迁移
└── docs/ # 文档
```
## 核心业务对象
### Domain Entities
**User** - 用户
- 基本信息:id, email, display_name
**Workspace** - 工作空间
- 用户的顶级组织单元
- 拥有者:owner_user_id
**Project** - 项目
- 属于某个 Workspace
- 包含多个 AssetLibrary
**AssetLibrary** - 素材库
- 类型:VIDEO(视频)/ VOICE(音频)
- 属于某个 Project
**Asset** - 素材
- 单个素材文件
- 属于某个 AssetLibrary
- 包含:storage_key, mime_type, metadata
**IngestJob** - 入库任务
- 状态:PENDING → PROCESSING → COMPLETED/FAILED
- 负责:文件上传后的元数据提取、Asset 创建
- 结果:result_asset_id
**ClassificationJob** - 分类任务
- 状态:PENDING → PROCESSING → COMPLETED/FAILED
- 负责:Asset 的自动分类
- 分类:scenic(风景), product(产品), person(人物), animal(动物), food(美食), tech(科技), sport(运动), music(音乐), other(其他)
- 结果:classification + confidence
## 已完成功能
### API 接口(6 组)
**健康检查**
- `GET /api/health` - 健康检查
**项目管理**
- `GET /api/projects?workspace_id=...` - 项目列表
- `POST /api/projects` - 创建项目
**素材库管理**
- `GET /api/asset-libraries?project_id=...` - 素材库列表
- `POST /api/asset-libraries` - 创建素材库
**素材管理**
- `GET /api/assets?library_id=...` - 素材列表
- `POST /api/assets` - 创建素材
**任务管理**
- `POST /api/ingest-jobs` - 提交入库任务
**文件上传**
- `POST /api/upload` - 上传素材文件
### Worker 任务(3 个)
**健康检查**
- `worker.healthcheck` - Worker 健康检查
**入库任务**
- `worker.ingest_asset` - 素材入库处理
- 元数据提取(当前 mock,真实场景用 ffprobe)
- Asset 创建
- IngestJob 状态更新
**分类任务**
- `worker.classify_asset` - 素材分类处理
- 自动分类(当前 mock,真实场景用 ML 模型或 vision API)
- ClassificationJob 状态更新
### 完整业务流程
**上传→入库→Asset 创建**
1. 用户通过 `POST /api/upload` 上传文件
2. API 生成 storage_key,创建 IngestJob(状态 PENDING)
3. Celery 任务 `ingest_asset` 被入队
4. Worker 处理:
- 更新状态为 PROCESSING
- 提取元数据
- 创建 Asset 实体
- 更新 IngestJob 状态为 COMPLETED,记录 result_asset_id
5. 异常时更新状态为 FAILED,记录 error_message
**分类→结果**
1. 创建 ClassificationJob(状态 PENDING)
2. Celery 任务 `classify_asset` 被入队
3. Worker 处理:
- 更新状态为 PROCESSING
- 运行分类模型
- 更新 ClassificationJob 状态为 COMPLETED,记录 classification + confidence
4. 异常时更新状态为 FAILED,记录 error_message
## 开发指南
### 环境准备
### 1. 克隆仓库
```bash
# 1. 安装依赖
git clone https://gitea.your-server.com/xiaoxia/xiaoxia-saas.git
cd xiaoxia-saas
```
### 2. 安装依赖
```bash
# Python 依赖
pip install -r requirements.txt
# 2. 启动开发环境(Docker Compose)
cd infra/docker
docker-compose up -d
# 3. 运行数据库迁移
alembic upgrade head
# 4. 启动 API(开发模式)
cd apps/api
uvicorn main:app --reload --host 0.0.0.0 --port 8000
# 5. 启动 Worker(开发模式)
cd apps/worker
celery -A worker_app.celery_app.celery_app worker --loglevel=info
# 或使用虚拟环境
python -m venv venv
source venv/bin/activate # Linux/Mac
# venv\Scripts\activate # Windows
pip install -r requirements.txt
```
### 运行测试
### 3. 配置环境变量
```bash
# 运行所有集成测试
pytest tests/integration/ -v
# 运行指定测试
pytest tests/integration/test_ingest_pipeline.py -v
# 运行所有测试(包含覆盖率)
pytest --cov=packages --cov=apps --cov-report=html
cp .env.example .env
# 编辑 .env 文件,填写实际配置
```
### 数据库迁移
**必须配置的项:**
- `DATABASE_URL` - PostgreSQL 连接字符串
- `JWT_SECRET_KEY` - JWT 密钥(生产环境必须修改)
- `SMTP_*` - 邮件服务配置(用于发送验证邮件)
### 4. 初始化数据库
```bash
# 创建新迁移
alembic revision -m "description"
# 创建数据库
psql -U postgres -c "CREATE DATABASE xiaoxia_saas;"
# 应用迁移
alembic upgrade head
# 回滚迁移
alembic downgrade -1
# 查看当前版本
alembic current
# 查看迁移历史
alembic history
# 执行迁移
psql $DATABASE_URL -f migrations/001_initial_schema.sql
```
## 下一步计划
### 5. 启动服务
### Phase 2 - 基础设施完善(进行中)
- [x] CI/CD 流水线(Gitea Actions)
- [ ] 真实文件存储(MinIO / S3)
- [ ] 生产环境配置(环境变量、密钥管理)
- [ ] 监控与日志(Prometheus + Grafana)
- [ ] API 文档(Swagger / ReDoc)
```bash
# 开发环境
uvicorn apps.api.main:app --reload --host 0.0.0.0 --port 8000
### Phase 3 - 核心业务扩展
- [ ] 用户认证与授权(JWT)
- [ ] Workspace 多用户协作
- [ ] 视频剪辑任务(ClipJob)
- [ ] 音频处理任务(AudioProcessJob)
- [ ] 任务队列管理与监控
- [ ] Webhook 通知
# 生产环境
uvicorn apps.api.main:app --host 0.0.0.0 --port 8000 --workers 4
```
### Phase 4 - 前端开发
- [ ] 用户登录/注册页面
- [ ] 工作空间管理
- [ ] 项目管理
- [ ] 素材库管理
- [ ] 素材上传与预览
- [ ] 任务状态监控
### 6. 访问 API 文档
### Phase 5 - 高级功能
- [ ] 真实 ML 模型集成(分类、识别)
- [ ] 批量处理
- [ ] 定时任务
- [ ] 数据分析与报表
- [ ] API 限流与配额
## 架构决策记录
### ADR-001: Clean Architecture
**日期**: 2026-06-15
**状态**: 已采纳
**决策**: 采用 Clean Architecture 重新设计系统
**原因**:
- 旧 desktop 系统耦合严重,难以测试和维护
- 新 SaaS 需要长期演进,架构需要可扩展
- Clean Architecture 提供清晰的依赖方向和边界
### ADR-002: 双持久化实现
**日期**: 2026-06-15
**状态**: 已采纳
**决策**: 同时提供 In-Memory 和 SQLAlchemy 两种 Repository 实现
**原因**:
- In-Memory 实现用于测试,快速且无外部依赖
- SQLAlchemy 实现用于生产,真实数据库持久化
- Repository Pattern 使得实现可随时切换
### ADR-003: 集成测试优先
**日期**: 2026-06-15
**状态**: 已采纳
**决策**: 集成测试优先于单元测试
**原因**:
- 核心业务流程需要端到端验证
- In-Memory 实现使得集成测试成本低
- 单元测试在架构稳定后逐步补充
## Commits 历史
1. `b5a62ee` - feat: initial SaaS scaffold
2. `43fd071` - feat: implement ingest asset worker task
3. `d550416` - feat: add upload asset endpoint
4. `e4e2595` - feat: add PostgreSQL persistence layer
5. `b43cdca` - feat: add Alembic database migrations
6. `a8177c1` - feat: add asset classification pipeline
## 贡献指南
### 代码风格
- 遵循 PEP 8
- 使用 Black 格式化代码
- 使用 type hints
- 中文注释与文档
### Commit 规范
- feat: 新功能
- fix: 修复
- docs: 文档
- test: 测试
- refactor: 重构
- chore: 构建/工具
### Pull Request
1. 基于 `main` 创建新分支
2. 编写测试并确保通过
3. 更新相关文档
4. 提交 PR,描述改动内容
## 许可证
内部项目,未公开。
## 联系方式
技术问题:请联系小虾 AI 团队
打开浏览器访问:
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
---
**最后更新**: 2026-06-15
**当前版本**: 0.1.0 (Phase 1 完成)
## 📖 API 使用示例
### 注册用户
```bash
curl -X POST http://localhost:8000/api/v1/auth/register \
-H "Content-Type: application/json" \
-d '{
"email": "user@example.com",
"password": "SecurePass123",
"username": "myusername",
"display_name": "My Name"
}'
```
### 登录
```bash
curl -X POST http://localhost:8000/api/v1/auth/login \
-H "Content-Type: application/json" \
-d '{
"email": "user@example.com",
"password": "SecurePass123"
}'
```
返回:
```json
{
"access_token": "eyJ0eXAiOiJKV1QiLCJhbGc...",
"refresh_token": "abc123...",
"token_type": "bearer",
"user_id": "...",
"expires_in": 1800
}
```
### 创建工作空间
```bash
curl -X POST http://localhost:8000/api/v1/workspaces \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
-d '{
"name": "我的工作空间",
"subscription_plan": "free"
}'
```
---
## 🧪 运行测试
```bash
# 运行所有单元测试
pytest tests/unit -v
# 运行集成测试
pytest tests/integration -v
# 生成覆盖率报告
pytest --cov=packages --cov-report=html
```
---
## 📁 项目结构
```
xiaoxia-saas/
├── apps/
│ └── api/ # FastAPI 应用
│ ├── main.py # 应用入口
│ └── app/
│ ├── api/routes/ # API 路由
│ ├── middleware/ # 中间件
│ └── dependencies.py # 依赖注入
├── packages/
│ ├── domain/ # 领域模型
│ ├── application/ # 用例层
│ ├── ports/ # 接口定义
│ └── adapters/ # 适配器实现
├── migrations/ # 数据库迁移
├── tests/ # 测试
└── docs/ # 文档
```
---
## 🔧 开发工具
### 代码格式化
```bash
# 安装工具
pip install black isort
# 格式化代码
black packages/ apps/ tests/
isort packages/ apps/ tests/
```
### 类型检查
```bash
pip install mypy
mypy packages/ apps/
```
---
## 🐳 Docker 部署
```bash
# 构建镜像
docker build -t xiaoxia-saas:latest .
# 运行容器
docker run -d \
--name xiaoxia-saas \
-p 8000:8000 \
--env-file .env \
xiaoxia-saas:latest
```
---
## 📚 更多文档
- [Phase 4 完成总结](docs/PHASE4-COMPLETE.md)
- [数据库迁移指南](migrations/README.md)
- [API 设计文档](docs/PHASE4-DESIGN.md)
---
## 🆘 常见问题
### Q: 邮件发送失败?
A: 检查 SMTP 配置,Gmail 需要使用应用专用密码。
### Q: 数据库连接失败?
A: 确认 PostgreSQL 正在运行,DATABASE_URL 配置正确。
### Q: JWT token 无效?
A: 检查 JWT_SECRET_KEY 是否配置,access_token 是否过期。
---
**支持联系:** xiaoxia@example.com
+56
View File
@@ -0,0 +1,56 @@
"""
应用配置
"""
import os
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
"""应用配置"""
# 应用配置
APP_NAME: str = "小虾 SaaS"
APP_VERSION: str = "1.0.0"
BASE_URL: str = "http://localhost:3000"
# 数据库配置
DATABASE_URL: str = "postgresql://xiaoxia:password@localhost:5432/xiaoxia_saas"
# Redis 配置
REDIS_URL: str = "redis://localhost:6379/0"
# JWT 配置
JWT_SECRET_KEY: str = "your-secret-key-change-in-production"
JWT_ALGORITHM: str = "HS256"
JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30
# 邮件配置
SMTP_HOST: str = "smtp.gmail.com"
SMTP_PORT: int = 587
SMTP_USER: str = ""
SMTP_PASSWORD: str = ""
SMTP_FROM_EMAIL: str = ""
SMTP_FROM_NAME: str = "小虾 SaaS"
# 环境
ENVIRONMENT: str = "development" # development, staging, production
DEBUG: bool = True
# CORS
CORS_ORIGINS: list = ["http://localhost:3000", "http://localhost:5173"]
class Config:
env_file = ".env"
case_sensitive = True
@lru_cache()
def get_settings() -> Settings:
"""获取配置(单例)"""
return Settings()
# 全局配置实例
settings = get_settings()
+54 -60
View File
@@ -1,67 +1,61 @@
"""
FastAPI 主应用
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from app.api.router import api_router
from app.core.config import get_settings
from apps.api.app.api.routes import api_router
# 创建 FastAPI 应用
app = FastAPI(
title="小虾 SaaS API",
description="自动化剪辑 SaaS 平台 API",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc",
)
# CORS 中间件
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost:3000",
"http://localhost:5173",
"https://yourdomain.com",
],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Gzip 压缩
app.add_middleware(GZipMiddleware, minimum_size=1000)
# 注册路由
app.include_router(api_router)
def create_app() -> FastAPI:
settings = get_settings()
app = FastAPI(
title="小虾 SaaS API",
description="""
小虾 SaaS 自动化剪辑系统 API
## 功能模块
### 📁 资源库管理
- **Projects**: 项目管理
- **Asset Libraries**: 资产库管理
- **Assets**: 素材资产管理
### 📤 素材导入
- **Upload**: 文件上传(支持 MinIO 对象存储)
- **Ingest Jobs**: 素材导入任务管理
### 🎬 自动化剪辑
- 智能场景分割
- 自动转场
- 字幕生成
## 技术栈
- FastAPI + Python 3.12
- PostgreSQL 数据库
- Redis 队列
- Celery 异步任务
- MinIO 对象存储
""",
version="0.1.0",
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json",
contact={
"name": "小虾团队",
"email": "dev@xiaoxiajianji.com",
},
license_info={
"name": "Proprietary",
},
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins_list,
allow_credentials=settings.cors_allow_credentials,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API routes
app.include_router(api_router, prefix=settings.api_prefix)
return app
@app.get("/")
async def root():
"""健康检查"""
return {
"service": "小虾 SaaS API",
"status": "running",
"version": "1.0.0",
}
app = create_app()
@app.get("/health")
async def health_check():
"""健康检查接口"""
return {
"status": "healthy",
"database": "ok", # TODO: 实际检查数据库连接
"redis": "ok", # TODO: 实际检查 Redis 连接
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)