XiaZheStudy 后端解读

后端用 FastAPI 写的，标准的三层架构：Router → Service → Model。看完这篇，你就知道后端代码是怎么组织的。

技术栈

目录结构

backend/
├── app/
│   ├── main.py              # 入口文件，FastAPI 实例
│   ├── api/
│   │   └── v1/
│   │       ├── __init__.py  # 路由注册
│   │       └── endpoints/   # 各个接口模块
│   │           ├── auth.py          # 认证相关
│   │           ├── users.py         # 用户管理
│   │           ├── documents.py     # 文档管理
│   │           ├── courses.py       # 课程管理
│   │           ├── ai.py            # AI 生成
│   │           ├── posts.py         # 广场帖子
│   │           ├── referrals.py     # 邀请系统
│   │           ├── messages.py      # 站内信
│   │           ├── export_tasks.py  # 导出任务
│   │           ├── upload.py        # 文件上传
│   │           └── activation_codes.py  # 激活码
│   ├── core/
│   │   ├── config.py        # 配置管理
│   │   ├── supabase_db.py   # 数据库连接
│   │   ├── dependencies.py  # 依赖注入
│   │   └── storage_init.py  # 存储初始化
│   ├── models/              # 数据模型（SQLAlchemy）
│   │   ├── user.py
│   │   ├── document.py
│   │   ├── course.py
│   │   ├── post.py
│   │   └── ...
│   ├── schemas/             # Pydantic 模型（请求/响应）
│   ├── services/            # 业务逻辑层
│   │   ├── ai_service.py
│   │   └── document_parser.py
│   └── utils/               # 工具函数
├── alembic/                 # 数据库迁移
├── requirements.txt         # 依赖包
├── Dockerfile              # Docker 配置
└── .env                    # 环境变量

核心文件解读

1. main.py - 入口文件

# 创建 FastAPI 实例
app = FastAPI(
    title=settings.APP_NAME,
    version=settings.APP_VERSION,
    lifespan=lifespan,  # 生命周期管理
)

# 配置 CORS（跨域）
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins_list,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(api_router, prefix="/api/v1")

关键点：

• lifespan 管理启动和关闭时的操作（连接数据库、初始化存储）
• CORS 配置允许前端跨域访问
• 所有 API 都在 /api/v1 前缀下

2. 路由注册 - api/v1/init.py

api_router = APIRouter()

# 注册各个模块的路由
api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(documents.router, prefix="/documents", tags=["documents"])
api_router.include_router(courses.router, prefix="/courses", tags=["courses"])
api_router.include_router(ai.router, prefix="/ai", tags=["ai"])
# ...

最终 API 路径：

• /api/v1/auth/login - 登录
• /api/v1/users/me - 获取当前用户
• /api/v1/documents - 文档列表
• /api/v1/courses - 课程列表
• /api/v1/ai/generate/text - AI 生成

数据模型解读

User 模型

class User(Base):
    __tablename__ = "users"

    id: Mapped[int]                    # 主键
    supabase_user_id: Mapped[str]      # Supabase Auth ID
    email: Mapped[str]                 # 邮箱
    username: Mapped[str]              # 用户名
    avatar_url: Mapped[Optional[str]]  # 头像

    # 订阅相关
    subscription_tier: Mapped[str]     # free/basic/plus/pro
    points_balance: Mapped[int]        # 积分余额（默认 500）

    # 邀请系统
    referral_code: Mapped[str]         # 我的邀请码
    referred_by_id: Mapped[int]        # 谁邀请了我

    # Stripe 集成
    stripe_customer_id: Mapped[str]    # Stripe 客户 ID

    # 关联关系
    documents: Mapped[List["Document"]]
    courses: Mapped[List["Course"]]
    wallet: Mapped["UserWallet"]

Course 模型

class Course(Base):
    __tablename__ = "courses"

    id: Mapped[int]
    user_id: Mapped[int]               # 所属用户
    document_id: Mapped[Optional[int]] # 关联文档（可选）

    # 课程信息
    title: Mapped[str]                 # 标题
    description: Mapped[str]           # 描述
    cover_image: Mapped[str]           # 封面图

    # 生成配置
    style: Mapped[str]                 # 风格：standard/humorous/academic
    difficulty: Mapped[str]            # 难度：easy/medium/hard

    # 内容
    content: Mapped[dict]              # JSON 格式的动画内容

    # 统计
    views_count: Mapped[int]           # 播放次数
    likes_count: Mapped[int]           # 点赞数

    # 状态
    status: Mapped[str]                # pending/processing/completed/failed
    is_public: Mapped[bool]            # 是否公开

API 接口解读

AI 生成接口（核心）

@router.post("/generate/document")
async def generate_from_document(
    request: Request,
    gen_request: GenerateFromDocumentRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    # 1. 获取文档
    document = await db.execute(
        select(Document).where(
            Document.id == gen_request.document_id,
            Document.user_id == current_user.id
        )
    )

    # 2. 解析文档内容
    content = await document_parser.parse_from_storage(document.file_url)

    # 3. 流式生成动画
    async def event_generator():
        async for chunk in ai_service.generate_course_content_stream(
            content=content,
            style=gen_request.style,
            difficulty=gen_request.difficulty
        ):
            yield chunk

    # 4. 返回 SSE 流
    return StreamingResponse(event_generator(), headers={
        "Content-Type": "text/event-stream"
    })

关键点：

• Depends(get_current_user) - 依赖注入，自动验证用户身份
• Depends(get_db) - 依赖注入，获取数据库会话
• StreamingResponse - 流式响应，实时输出 AI 生成的内容
• SSE (Server-Sent Events) - 服务端推送，前端实时接收

依赖注入示例

# core/dependencies.py

async def get_current_user(
    request: Request,
    db: AsyncSession = Depends(get_db)
) -> User:
    # 从请求头获取 token
    token = request.headers.get("Authorization")
    if not token:
        raise HTTPException(status_code=401, detail="Not authenticated")

    # 验证 token，获取用户
    user = await verify_token_and_get_user(token, db)
    return user

async def get_db():
    async with AsyncSession(engine) as session:
        yield session

业务流程解读

文档上传流程

1. 前端调用 POST /api/v1/upload
2. 后端接收文件
3. 上传到 Supabase Storage
4. 创建 Document 记录
5. 返回文档信息

AI 生成流程

1. 前端调用 POST /api/v1/ai/generate/document
2. 后端验证用户身份
3. 获取文档，解析内容
4. 调用 AI 服务，流式生成
5. SSE 实时推送到前端
6. 前端渲染动画
7. 用户确认后保存为 Course

课程保存流程

1. 前端调用 POST /api/v1/courses
2. 后端创建 Course 记录
3. 保存动画内容（JSON）
4. 返回课程 ID

关键配置

环境变量

# 数据库（使用 Pooler 连接，避免 IPv6 问题）
DATABASE_URL=postgresql+asyncpg://postgres.xxx:[email protected]:5432/postgres

# Supabase
SUPABASE_URL=https://xxx.supabase.co
SUPABASE_KEY=eyJxxx...
SUPABASE_SERVICE_KEY=eyJxxx...

# CORS
CORS_ORIGINS=https://www.xiazhestudy.com,https://xiazhestudy.com

# JWT
SECRET_KEY=your-secret-key

# AI 服务
AI_API_KEY=your-ai-api-key

CORS 配置

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins_list,  # 允许的前端域名
    allow_credentials=True,                     # 允许携带 cookie
    allow_methods=["*"],                        # 允许所有 HTTP 方法
    allow_headers=["*"],                        # 允许所有请求头
)

代码规范总结

目录组织

endpoints/   - 接口定义（薄，只做请求处理）
services/    - 业务逻辑（厚，核心逻辑在这）
models/      - 数据模型（ORM 定义）
schemas/     - 请求响应模型（Pydantic）
core/        - 核心配置（数据库、依赖注入）
utils/       - 工具函数

命名规范

• 文件名：小写 + 下划线（user_wallet.py）
• 类名：大驼峰（UserWallet）
• 函数名：小写 + 下划线（get_current_user）
• 路由：RESTful 风格（/users, /users/{id}）

异步编程

• 所有数据库操作用 async/await
• 使用 AsyncSession 而不是 Session
• AI 生成使用 async for 流式输出

小结

这个后端项目是标准的 FastAPI 三层架构：

1. Router 层 - 接收请求，参数校验，调用 Service
2. Service 层 - 业务逻辑，调用 Model
3. Model 层 - 数据库操作

亮点：

• 完整的依赖注入
• SSE 流式响应
• 异步数据库操作
• 清晰的目录结构

适合作为 FastAPI 项目的参考模板。