本章目录

Telegram 机器人实战 › 第 4 部分 · 会让你翻车的地方

第 5 章

重启一下，数据全没了

哪些数据丢得起、哪些丢不起，是两个完全不同的问题。顺便还上第 3 章欠的时区债——它会让国外用户莫名其妙签不上到。

约 22 分钟含一个我自己踩过的坑aiogram 3.x · 最后验证 2026-08

先分清哪些丢得起

上一章结尾那个测试——重启之后兑换到一半的人卡住了——很容易让人得出「所有东西都得存起来」的结论。那是过度反应。

先把数据分成两类：

	是什么	丢了会怎样
丢不起	积分、签到记录、兑换记录	真金白银的损失。用户攒了三个月的分没了，他不会再回来。
丢得起	「他正在选商品」这种半截对话状态	用户重发一次 `/redeem` 就好了。整个流程也就十几秒。

第 3 章开始，第一类就已经在 SQLite 里了。这一章主要处理它的几个坑，顺便说清第二类什么时候才真的需要存。

半截对话：多数情况下真不用存

算一下概率：兑换流程从开始到结束十几秒，你一天重启几次？正好卡住某个人的概率低到可以忽略，而且代价只是他重来一次。

所以先别急着装 Redis。什么时候才真的需要持久化状态：

流程很长	比如一个要填十几项的报名表，用户可能填到一半去吃饭了
你要跑多个进程	两个进程各有各的内存，用户第一句话被 A 接了、第二句被 B 接了，状态对不上
部署很频繁	一天推十几次代码，那确实会烦到人

真到了这一步，aiogram 自带 Redis 支持，改动就两行：

bot.py · 需要时才换

from aiogram.fsm.storage.redis import RedisStoragestorage_fsm = RedisStorage.from_url("redis://localhost:6379/0")dp = Dispatcher(storage=storage_fsm)

代价是多一个要装、要监控、要备份的组件（pip install redis 再起一个 Redis）。没到那三种情况就先别背这个包袱。

✓ 一个折中做法

如果只是不想让重启后的用户一脸懵，加个兜底 handler 就够了：捕获那些「不在任何状态、又不是命令」的消息，回一句「我没听懂，试试 /start」。成本几行代码，解决八成尴尬。

还第 3 章欠的债：时区

第 3 章结尾列过一条：群里有国外的人，签到会出错。现在还它。

问题出在这一行：

handlers.py 里的老写法

day=date.today().isoformat()   # 用的是服务器时区

服务器一般跑在 UTC。实测给你看差多少：

	值
用户本地时间	`2026-08-21 07:00 CST`（他觉得是 21 号早上）
服务器同一时刻	`2026-08-20 23:00 UTC`
`date.today()` 算出	`2026-08-20`
结果	他 21 号早上签到，被记成 20 号。如果他 20 号晚上签过了，就会看到「今天已经签过了」——而他明明是新的一天。

反过来也有：住美西的人晚上签到，可能被算成第二天，于是一天签了两次。

修法：把时区跟着用户走

先给 users 表加一列。注意这里有个坑——建表语句改了不管用：

storage.py · 这样改是错的

CREATE TABLE IF NOT EXISTS users (    user_id INTEGER PRIMARY KEY,    name    TEXT    NOT NULL,    points  INTEGER NOT NULL DEFAULT 0,    tz      TEXT    NOT NULL DEFAULT 'Asia/Shanghai'   # ← 加了也没用)

IF NOT EXISTS 的意思是表已经在了就啥也不干——它不会去比对你改了什么。你本地删掉 data.db 重跑一切正常，线上老库照样没有这一列，一查就炸。

⬤ 这个坑我自己刚踩过

给这个网站的反馈功能加字段时，本地测得好好的，一部署到服务器就 502。日志里躺着一句 table feedback has no column named flag —— 生产库是老版本建的，多出来的那一列根本不存在。
本地能跑线上挂，这是最典型的一种。

正确做法是显式补列：

storage.py · init_db 里加这一段

def init_db() -> None:    with connect() as conn:        # ... 原来的 CREATE TABLE 不动 ...        # 老库补列：CREATE TABLE IF NOT EXISTS 不会给已存在的表加字段，        # 所以以后每加一列，都要在这里补一句。        cols = {row[1] for row in conn.execute("PRAGMA table_info(users)")}        for name, ddl in (            ("tz", "TEXT NOT NULL DEFAULT 'Asia/Shanghai'"),        ):            if name not in cols:                conn.execute(f"ALTER TABLE users ADD COLUMN {name} {ddl}")

然后算日期的时候按用户自己的时区来：

storage.py 追加

from datetime import datetimefrom zoneinfo import ZoneInfoDEFAULT_TZ = "Asia/Shanghai"def get_tz(user_id: int) -> str:    with connect() as conn:        row = conn.execute(            "SELECT tz FROM users WHERE user_id = ?", (user_id,)        ).fetchone()    return row[0] if row else DEFAULT_TZdef set_tz(user_id: int, tz: str) -> bool:    try:        ZoneInfo(tz)              # 先验证，别让乱输的字符串进库    except Exception:        return False    with connect() as conn:        conn.execute(            "INSERT INTO users (user_id, name, tz) VALUES (?, '', ?) "            "ON CONFLICT(user_id) DO UPDATE SET tz = excluded.tz",            (user_id, tz),        )    return Truedef today_for(user_id: int) -> str:    """按这个用户所在时区，算出「今天」是哪天。"""    return datetime.now(ZoneInfo(get_tz(user_id))).date().isoformat()

签到那里把 date.today().isoformat() 换成 storage.today_for(user_id) 就行。再加个命令让用户自己设：

handlers.py 追加

@router.message(Command("timezone"))async def cmd_timezone(message: Message, command: CommandObject) -> None:    if message.from_user is None:        return    arg = (command.args or "").strip()    if not arg:        cur = storage.get_tz(message.from_user.id)        await message.answer(            f"你现在的时区是 {cur}\n"            f"要改就发：/timezone Asia/Tokyo"        )        return    if not storage.set_tz(message.from_user.id, arg):        await message.answer(            "没这个时区。要写成 Asia/Shanghai 这种格式，"            "「北京」「GMT+8」都不行。"        )        return    await message.answer(f"好了，以后按 {arg} 算日期。今天是 "                         f"{storage.today_for(message.from_user.id)}")

CommandObject 从 aiogram.filters 导入，它能拿到命令后面跟的参数。

◆ 时区名必须是 IANA 格式

Asia/Shanghai 有效，北京、GMT+8、Asia/Beijing 全都无效（最后那个看着很像，其实不存在）。所以 set_tz 里那句 ZoneInfo(tz) 验证不能省——否则脏数据进了库，以后每次读都抛异常。

SQLite 的三个实际问题

一、连接不要开着不放

第 3 章那个 connect() 每次用完就关。看着浪费，其实是对的——SQLite 的连接不能跨线程共享，而 aiogram 是异步的，你分不清哪段代码跑在哪。每次开关的开销在这个量级下可以忽略。

二、默认锁等待只有 5 秒

两个写操作撞上时，后来的会等前一个提交。等超过 5 秒就抛 database is locked。签到这种小写入几乎撞不上，但加个超时更保险：

storage.py

conn = sqlite3.connect(DB_PATH, timeout=10)   # 默认 5 秒，给宽点

三、备份不能直接 cp

程序正在写的时候复制文件，可能拷到一个写了一半的库。用 SQLite 自己的备份接口，它会处理好一致性：

backup.py

import sqlite3from datetime import datetimefrom pathlib import Pathsrc = sqlite3.connect("data.db")name = f"backup-{datetime.now():%Y%m%d-%H%M}.db"dst = sqlite3.connect(name)src.backup(dst)          # 热备份，程序在跑也没事dst.close(); src.close()print(f"备份好了：{name}")

挂个 cron 每天跑一次就够了。没有备份的数据等于随时会消失的数据——这条不用等出事才信。

什么时候 SQLite 就不够了

网上很多人一上来就劝你用 PostgreSQL。对签到 bot 这种量级，那是白背负担。真正该换的信号只有这几个：

信号	为什么 SQLite 顶不住
要跑在多台机器上	SQLite 是一个文件。两台机器各读各的文件，数据就是两份。这一条最硬，没法绕。
持续每秒几百次写	SQLite 同一时刻只允许一个写入。读可以并发，写会排队。
需要全文搜索、复杂分析	SQLite 能做（FTS5 挺好用），但真要做数据分析就该换了
几千人的群、每天几千次操作	SQLite 完全够。别提前迁移。

第 3 章把数据操作单独放 storage.py，就是为了这一天：真要换，只改这一个文件，handlers 一行都不用动。

验一遍

怎么试	应该看到
用老的 `data.db` 直接启动	正常起来，不报 no such column —— 补列逻辑生效了
`/timezone`	显示当前时区
`/timezone Asia/Tokyo`	设置成功，并告诉你东京的今天是几号
`/timezone 北京`	提示格式不对，库里的值没被改坏
设成 `Pacific/Kiritimati`（UTC+14）再签到	能签上 —— 因为对他来说已经是新的一天了
跑一次 `backup.py`	生成一个带时间戳的备份文件，用 `sqlite3 备份文件 ".tables"` 能打开

这一章的要点

先分清丢得起和丢不起。半截对话状态多数情况真不用存，别为它装 Redis
CREATE TABLE IF NOT EXISTS 不会给老表加列 —— 本地删库重跑一切正常，线上直接炸。要显式 ALTER TABLE
日期要按用户时区算，date.today() 用的是服务器时区
时区名只认 IANA 格式，存之前必须验证
备份用 src.backup(dst)，别直接 cp
几千人的群 SQLite 完全够，真正该换的信号是「要跑在多台机器上」

✓ 试着改一个

首次签到的人自动猜时区：Telegram 的 message.from_user.language_code 能拿到语言（zh-hans、ja、en），做个粗略映射当默认值，再提示他「猜的，不对就 /timezone 改」。
猜错不要紧，要紧的是别默默猜错 —— 一定要告诉用户你猜了。

有没有踩到别的坑？

加字段、改时区、备份——这几件事上你遇到什么问题，说出来我补进这一章。留不留联系方式都行。

两个人同时抢最后一个名额

继续 →