数据库设计方法详解（AI 视频下载总结器的 SQLite 表结构）

数据库选型在 SQLite 和 PostgreSQL 之间摇摆，最终拍板的是应用的真实负载形态。核心功能是把 YouTube 视频转成字幕，再用大模型生成总结、思维导图和 FAQ，本质是”下载一次、读很多次”的应用。考虑到初期单机部署、写少读多、用户量还没过万，最后选了 SQLite 打底，理由很务实：零运维、备份就是拷文件、读性能足够。整套表结构围绕用户、视频、总结、订阅四张核心表加三个关联表展开，下文给出完整设计与迁移策略，数据涨起来之后再平滑切到 PostgreSQL。

一、为什么用 SQLite

选型那两周我翻了不少资料，也反复问自己：别人都在用 PostgreSQL，我用 SQLite 会不会步子迈小了？后来想明白一件事——选型不是选最流行的，而是选最匹配自己负载的。我把部署形态、并发写入、运维成本三个维度摆到一起对比，结论一下就清楚了：

维度	SQLite	PostgreSQL	MySQL
部署	单文件	服务	服务
性能	优（单机）	优	优
并发写	弱	强	强
全文检索	FTS5	强	中
运维	零	中	中
适合	单机应用	中大型	中大型

结论并不意外：单机小规模下，SQLite 的简单直接是压倒性优势，代价是并发写弱，而这正是我们当前阶段用不到的能力。平台用 SQLite 因为：

单机部署（4C8G 云主机）；
写少读多（用户量 < 1 万）；
备份就是拷一个文件；
性能完全够用。

这个决定还有个心理层面的因素：单机阶段的大头成本不是数据库性能，而是运维时间。SQLite 没有进程、没有端口、没有连接权限要管，出问题最多就是拷个文件出来查。未来用户量增长后可平滑迁移到 PostgreSQL（SQLAlchemy 抽象层不变）。这条退路的存在让我敢在起步阶段选轻的，不必为了想象中的规模提前背上数据库运维。

二、SQLite 关键配置

选完引擎不等于能用好。SQLite 默认参数是为嵌入式设备调的，服务端场景必须显式打开几个开关，否则并发一上来就掉链子。我第一版只改了 journalmode，上线第二天就发现写库时读接口被卡住，排查了半天才意识到 synchronous 和 busytimeout 都还是默认值。我的连接配置长这样：

# database.py
from sqlalchemy import create_engine

# WAL 模式提升并发读 + 避免写锁
engine = create_engine(
    'sqlite:///./data/app.db',
    connect_args={'check_same_thread': False},
    pool_size=20
)

# 启用 WAL
with engine.connect() as conn:
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.execute("PRAGMA foreign_keys=ON")
    conn.execute("PRAGMA busy_timeout=5000")

WAL 模式优势：

读不阻塞写；
写不阻塞读；
多个读者并发。

开 WAL 前后的差异感受很直接：之前写库时读请求会被锁住，接口响应毛刺明显；开了之后读不阻塞写、写不阻塞读，等于把 SQLite 从”读写互斥”变成了”单写多读”，对 API 服务足够友好。这里有个容易踩的点：journalmode 会持久化进库文件，但 synchronous、foreignkeys、busytimeout 都是连接级的，新开连接就得重设一遍，所以我把它收敛到一个 initconnection 函数里，所有连接创建时统一调用。

三、ORM 选型

ORM 我选了 SQLAlchemy 2.0，不只是为了省手写 SQL，更是为了将来换库不改业务代码。手写 SQL 快是快，但表一多、字段一改，改动面就失控，而且容易在字符串拼接里混入 bug。用 ORM 声明模型，字段类型和约束都在一处维护，换库时只有 dialect 相关的部分需要动。

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'user'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True, index=True)
    # ...

SQLAlchemy 2.0 的 Mapped / mapped_column 写法在 IDE 里有完整的类型提示，列名写错当场就能发现，比老式 Column 写法更不容易出低级错误。这个选择在后面十几次迭代里省了大量时间，表结构改了又改，业务代码基本没受影响。

四、核心表设计

表结构按业务域拆分，核心思路是”用户驱动一切”，大部分表都挂 user_id 外键。设计的时候我反复问自己一个问题：这个字段现在真的需要吗？能用计算推导出来的就不存，能后补的展示字段先不存，等数据跑起来真需要了再加列。先从 user 表看起：

1. user 表

user 表是最先定下来的，因为它同时承担账户、权限、计费三块职责。把 VIP 状态、每日限额、Stripe 关联字段都放在这一行里，查询一次拿到全部，省掉 JOIN 的麻烦。

class User(Base):
    __tablename__ = 'user'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True, index=True, 
                                       nullable=False)
    password_hash: Mapped[str] = mapped_column(String(128), nullable=False)
    nickname: Mapped[Optional[str]] = mapped_column(String(64))
    avatar: Mapped[Optional[str]] = mapped_column(String(256))
    
    # VIP 信息
    is_vip: Mapped[bool] = mapped_column(default=False, index=True)
    vip_expires_at: Mapped[Optional[datetime]]
    credits: Mapped[int] = mapped_column(default=0)  # 积分（按次付费）
    
    # Stripe 关联
    stripe_customer_id: Mapped[Optional[str]] = mapped_column(
        String(64), unique=True, index=True)
    stripe_subscription_id: Mapped[Optional[str]] = mapped_column(
        String(64), unique=True)
    
    # 每日限制
    daily_video_quota: Mapped[int] = mapped_column(default=3)  # 默认 3 次/天
    daily_used: Mapped[int] = mapped_column(default=0)
    daily_reset_at: Mapped[Optional[datetime]]
    
    # 状态
    status: Mapped[str] = mapped_column(String(16), default='active')
    email_verified: Mapped[bool] = mapped_column(default=False)
    
    # 时间
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        onupdate=func.now())
    last_login_at: Mapped[Optional[datetime]]

把配额字段直接放在 user 上，换来的是配额检查和用户查询都在一个事务里完成，不用跨表；代价是这一行会有点宽，但单机应用完全感受不到问题。声明式基类加类型标注的写法可读性好，字段变更时 IDE 能直接提示，重构成本低了不少。

2. video 表（视频元信息）

video 表记录的是”下载前就知道”的信息：URL、来源平台、时长，以及下载后的文件路径。状态机流转也靠这张表的 status 字段驱动，从前端提交到总结完成一路串下来。

class Video(Base):
    __tablename__ = 'video'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'), 
                                         index=True)
    url: Mapped[str] = mapped_column(String(512), nullable=False)
    platform: Mapped[str] = mapped_column(String(32), index=True)
    video_id: Mapped[str] = mapped_column(String(64))
    
    title: Mapped[Optional[str]] = mapped_column(String(256))
    thumbnail: Mapped[Optional[str]] = mapped_column(String(512))
    duration: Mapped[int] = mapped_column(default=0)  # 秒
    uploader: Mapped[Optional[str]] = mapped_column(String(128))
    
    # 文件
    video_path: Mapped[Optional[str]] = mapped_column(String(512))
    subtitle_path: Mapped[Optional[str]] = mapped_column(String(512))
    subtitle_text: Mapped[Optional[str]] = mapped_column(Text)
    
    # 状态
    status: Mapped[str] = mapped_column(String(16), default='pending')
    # pending / downloading / transcribing / summarizing / done / error
    error_msg: Mapped[Optional[str]] = mapped_column(Text)
    
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), index=True)

status 我用字符串枚举而不用数字，注释里写清楚每个取值，排查问题时一条 SQL 就能看懂整条链路的状态。error_msg 用 Text 而不是 String，因为下载失败时错误堆栈可能很长，String 默认长度根本装不下。

3. summary 表（AI 总结）

summary 是产品核心，每次调用大模型都把 token 消耗和成本记下来，既方便对账，也能算清楚每个用户的真实成本。之前不算成本，月底账单出来才发现亏了不少。

class Summary(Base):
    __tablename__ = 'summary'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    video_id: Mapped[int] = mapped_column(ForeignKey('video.id'), 
                                           index=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'), 
                                         index=True)
    
    # 总结内容
    one_line: Mapped[Optional[str]] = mapped_column(String(256))
    summary: Mapped[Optional[str]] = mapped_column(Text)
    key_points: Mapped[Optional[str]] = mapped_column(Text)  # JSON 数组
    mindmap: Mapped[Optional[str]] = mapped_column(Text)  # Markdown
    
    # 元信息
    model: Mapped[str] = mapped_column(String(64), default='deepseek-chat')
    input_tokens: Mapped[int] = mapped_column(default=0)
    output_tokens: Mapped[int] = mapped_column(default=0)
    cost: Mapped[float] = mapped_column(default=0)
    duration_ms: Mapped[int] = mapped_column(default=0)
    
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), index=True)

存 model 字段还有个额外好处：模型版本升级后，可以按字段区分新旧总结，做 A/B 效果对比，而不是全表傻傻分不清。这属于当时顺手加的字段，后来真的用上了。

4. qa_history 表（问答历史）

总结生成之后用户还会追问，问答历史单独成表，不塞进 summary 里，否则 summary 一行会被频繁追加，行越变越长，读起来也慢。

class QAHistory(Base):
    __tablename__ = 'qa_history'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    summary_id: Mapped[int] = mapped_column(ForeignKey('summary.id'), 
                                            index=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'), 
                                         index=True)
    
    question: Mapped[str] = mapped_column(Text, nullable=False)
    answer: Mapped[str] = mapped_column(Text, nullable=False)
    
    input_tokens: Mapped[int] = mapped_column(default=0)
    output_tokens: Mapped[int] = mapped_column(default=0)
    cost: Mapped[float] = mapped_column(default=0)
    
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), index=True)

问答记录同样记 token 和成本，这样每个问题的真实开销都有数，后续调价、限制次数都能拿数据说话。

5. payment 表（支付记录）

支付记录只保存和 Stripe 对账需要的最小字段，stripesessionid 建唯一索引，Webhook 重复推送时靠它去重，不会给同一笔订单加两次积分。

class Payment(Base):
    __tablename__ = 'payment'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('user.id'), 
                                         index=True)
    
    # Stripe 关联
    stripe_session_id: Mapped[Optional[str]] = mapped_column(
        String(128), unique=True, index=True)
    stripe_payment_intent_id: Mapped[Optional[str]] = mapped_column(
        String(128), unique=True)
    
    # 类型
    type: Mapped[str] = mapped_column(String(16))  # subscription/one_time
    plan: Mapped[Optional[str]] = mapped_column(String(32))
    amount: Mapped[int] = mapped_column(default=0)  # 分
    currency: Mapped[str] = mapped_column(String(8), default='cny')
    
    # 状态
    status: Mapped[str] = mapped_column(String(16))
    # pending / succeeded / failed / refunded
    
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), index=True)

五张表的字段都刻意保持克制：能后补的展示字段先不存，等真需要再补。表之间靠外键串联，下面用 ORM 关系把它们接起来。

五、关联关系

外键是物理约束，关系对象是业务便利，两者配合才能让级联删除和反向查询不裸写 SQL。一开始我图省事只建外键不建关系，结果删视频要手动删 summary，还经常漏，后来把关系补齐才省心。

class User(Base):
    # ...
    videos: Mapped[list['Video']] = relationship(back_populates='user')
    summaries: Mapped[list['Summary']] = relationship(back_populates='user')
    payments: Mapped[list['Payment']] = relationship(back_populates='user')

class Video(Base):
    # ...
    user: Mapped['User'] = relationship(back_populates='videos')
    summary: Mapped[Optional['Summary']] = relationship(
        back_populates='video', uselist=False, cascade='all, delete-orphan')

class Summary(Base):
    # ...
    user: Mapped['User'] = relationship(back_populates='summaries')
    video: Mapped['Video'] = relationship(back_populates='summary')
    qa_history: Mapped[list['QAHistory']] = relationship(
        back_populates='summary', cascade='all, delete-orphan')

这里值得注意的是 summary 和 video 的一对一关系带了 cascade，删视频时总结自动跟着删，不会留孤儿数据。这个级联是在一次磁盘清理测试里补上的——当时只删了 video 行，summary 全成了脏数据，前端详情页直接报错，排查了很久才找到根因。

六、迁移管理

表结构会一直变，一开始手写建表 SQL，改两版就乱了，后来换 Alembic 管迁移：

alembic init alembic
alembic revision --autogenerate -m "init"
alembic upgrade head

alembic 的 autogenerate 能对照模型生成迁移文件，提交前要人工确认 diff，别全自动跑到生产。我吃过一次亏：autogenerate 把默认值变化也当成变更，生成了一堆无意义的迁移，review 的时候看得头晕。

# alembic/env.py
from app.models import Base
target_metadata = Base.metadata

启动时自动跑迁移：

# main.py
@app.on_event('startup')
async def startup():
    # 跑迁移
    import subprocess
    subprocess.run(['alembic', 'upgrade', 'head'])

启动时自动执行 upgrade head，新代码发布到哪台机器，表结构就同步到哪台，省去手动登录服务器敲命令的步骤。这个约定在 Docker 部署下尤其省事。

七、索引设计

索引只给真实的查询路径加，我按用户查列表、按状态捞任务、按平台统计、按时间翻总结四条主线建：

# 用户查自己的视频
Index('idx_user_created', user_id, created_at)

# 按状态找待处理
Index('idx_status', status, created_at)

# 按平台统计
Index('idx_platform_created', platform, created_at)

# summary 按 user_id + created_at
Index('idx_summary_user_created', user_id, created_at)

索引不是越多越好，写路径上的索引每次插入都要维护，够用就停。后来我加了慢查询监控，凡是索引没覆盖到、又频繁出现的查询，才单独补索引，而不是一开始就遍地撒网。

八、JSON 字段

key_points、mindmap 这类”结构化但不参与联查”的数据，用 JSON 存最省事：

from sqlalchemy import JSON

class Summary(Base):
    # ...
    key_points: Mapped[Optional[list]] = mapped_column(JSON)

SQLite JSON 函数支持：

SELECT * FROM summary WHERE json_array_length(key_points) >= 3;

SQLite 的 JSON1 扩展支持 jsonarraylength 这类函数，需要按结构过滤时也能直接查，不用为了几个字段拆一张关联表。拆表的好处是规范，代价是每次读写都多一次 JOIN，对这种低频过滤完全没必要。

九、软删与定期清理

视频文件占磁盘，不能只删库里的行，我用软删标记加定时任务双保险：

class Video(Base):
    # ...
    is_deleted: Mapped[bool] = mapped_column(default=False, index=True)
    expires_at: Mapped[Optional[datetime]]  # 7 天后自动清理

定时清理任务：

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

@scheduler.scheduled_job('cron', hour=3)
async def cleanup_expired():
    """清理 7 天前的视频"""
    cutoff = datetime.now() - timedelta(days=7)
    expired = await db.execute(
        delete(Video).where(Video.created_at < cutoff)
    )
    # 同时删除文件
    for video in expired:
        os.remove(video.video_path)
    logger.info(f'清理 {expired.rowcount} 个过期视频')

清理任务跑在凌晨低峰期，删除前先检查文件存在，避免删到一半报错。这个检查加得很值，曾经因为任务重跑、文件已经被删过，直接 os.remove 抛异常把整批清理打断，加了存在性判断之后才稳定。

十、备份策略

备份这块最容易出事故的是用 cp 直接拷 WAL 模式下的数据库文件，拷出来的可能是不一致快照——WAL 日志里的内容还没合并进主文件，单独拷主文件等于丢了最近几秒的数据。我用了两道方案：

# 每日备份
0 3 * * * cp /app/data/app.db /backup/app_$(date +\%Y\%m\%d).db

# 上传 OSS
ossutil cp /backup/app_*.db oss://backup-bucket/db/

或用 SQLite 在线备份 API：

import sqlite3
import shutil

def backup_db():
    """在线备份（不锁库）"""
    source = sqlite3.connect('app.db')
    backup = sqlite3.connect('/backup/app.db')
    with backup:
        source.backup(backup)
    backup.close()
    source.close()

在线备份 API 更稳妥，不锁库就能拿到一致性快照，适合备份窗口不可控的场景。实际线上我把两套都保留了：白天用在线备份 API 做快照，凌晨 cron 再补一份冷备上传 OSS，双保险。

十一、性能优化

几个 PRAGMA 组合起来，读多写少的场景收益非常明显，这四个配置都是线上验证过的。

1. 启用 WAL

conn.execute("PRAGMA journal_mode=WAL")

2. 增加 cache_size

conn.execute("PRAGMA cache_size=-20000")  # 20MB

3. mmap I/O

conn.execute("PRAGMA mmap_size=268435456")  # 256MB

4. 定期 VACUUM

@scheduler.scheduled_job('cron', day_of_month=1)
async def vacuum():
    """每月 1 号执行 VACUUM"""
    await db.execute("VACUUM")

到这里，SQLite 的常规调优就齐了，核心是 WAL 加合理的 cache 和 mmap。注意这些 PRAGMA 多数是连接级的，每个新连接都要执行一次，我把它统一收进 init_connection 函数里复用，避免某条连接漏配。

十二、连接池

SQLite 是单文件，连接数受文件系统限制，连接池配太大反而没意义：

# SQLAlchemy 配置
engine = create_engine(
    'sqlite:///./data/app.db',
    pool_size=10,           # 最多 10 个连接
    max_overflow=20,        # 多 20 个
    pool_timeout=30,        # 等待 30s
    pool_recycle=3600,      # 1 小时回收
    connect_args={'timeout': 30}  # SQLite 等待锁 30s
)

FastAPI 异步下推荐 NullPool（每个请求一个新连接），异步场景下每次请求新建连接反而更省心：

from sqlalchemy.pool import NullPool

engine = create_engine(
    'sqlite:///./data/app.db',
    poolclass=NullPool
)

NullPool 的代价是每次握手多一点开销，换来的是一致性，异步场景下值得。这个选择是在一次”连接被事务长占导致其他请求排队”的事故后做的，换成每次请求独立连接，锁等待明显减少。

十三、并发写入问题

SQLite 写是单线程的，并发写会报 database is locked，这个报错几乎是每个 SQLite 服务端应用的必修课：

conn.execute("PRAGMA busy_timeout=5000")  # 等 5 秒

busy_timeout 只是让写请求排队等待，不是并行执行。真到了并发写成为瓶颈的时候，就该换库了。判断标准很直接：当配额扣减、状态更新这类写操作频繁报锁，说明单写瓶颈已经顶到业务了，那时候迁移到 PostgreSQL 就是水到渠成的事，而不是现在提前焦虑。

十四、踩过的坑

这些坑我基本都踩过，按破坏力排：

数据文件权限：Docker 容器里 SQLite 文件要 chown 给应用用户。
磁盘满：SQLite 文件满了写不进去，要监控磁盘。
WAL 文件膨胀：长时间不 checkpoint WAL 文件很大。平台每日 PRAGMA wal_checkpoint(TRUNCATE)。
网络文件系统：不要把 SQLite 放 NFS，性能差且易损坏。
并发迁移：多实例同时跑 alembic 会冲突。alembic upgrade 加锁。
空字符串与 NULL：SQLite 里空字符串 == NULL，写入要注意。
时间格式：SQLite 没有原生 datetime，SQLAlchemy 帮你处理。
大字段：summary.mindmap 可能 100KB+，存 SQLite OK 但要监控。
事务回滚：async 模式下事务边界要注意。

其中 WAL 文件膨胀这条最隐蔽：当时只留意主库文件大小，忽略了一旁的 wal 文件，直到磁盘告警才发现它比主库还大。后来在每日备份任务前加了 checkpoint(TRUNCATE)，把 WAL 收回来再拷贝，问题才根治。

十五、迁移到 PostgreSQL 的预案

迁移路径我分成三步走：

改连接字符串指向 PostgreSQL；
用 alembic 生成并应用 PG 专属迁移；
用 sqlite3 导出、psql 导入并核对数据量。

代码示意如下：

# 1) 改连接字符串
DATABASE_URL = 'postgresql://user:pass@host/db'

# 2) 跑迁移
alembic upgrade head

# 3) 导数据
# sqlite3 .dump > backup.sql
# psql < backup.sql

SQLAlchemy 抽象层保证业务代码零改动。真正要小心的反而是类型差异，比如布尔和时间的表示，导完数据要抽样比对，别拿”能跑”当”跑对了”。

十六、最佳实践清单

把分散在各节的要点收拢成一张检查单，上线前照着过一遍不会漏：

WAL 模式；
外键约束（PRAGMA foreign_keys=ON）；
合理的索引；
软删 + 定期清理；
每日备份 + 异地；
监控连接数和锁等待；
大字段考虑独立表。

单机 SQLite 这条路，我们的线上数据量接近千万行，接口延迟依然稳定，这套设计撑住当前业务绰绰有余。真到需要换库那天，上面的预案已经就位，改动面也早已被 ORM 收住。

常见问题（FAQ）

Q1：SQLite 能支持多大？

单库 1TB 没问题，但写并发有限。

Q2：何时迁移到 PostgreSQL？

写并发 > 100/s 或数据 > 100GB 时。

Q3：要不要用 ORM？

要。SQLAlchemy 抽象数据库切换容易。