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

    第 8 章

    到期该踢人，但他刚好在续费

    让机器人自己动起来：定时扫描、到期处理、每日推送。麻烦在于定时任务和用户操作会撞车——我实测过，付了钱的人真的会被踢出去。

    约 21 分钟让机器人自己动aiogram 3.x · 最后验证 2026-08
    01

    到目前为止，都是用户先开口

    前七章的所有代码都是被动的：用户发一条消息，你回一条。但很多功能需要机器人自己动：

    会员到期了把人移出群，或者提醒他续费
    每天早上八点推送昨天的签到排行
    连续三天没签到发个提醒
    每天凌晨备份数据库、清理过期数据

    这类活儿没有触发它的消息，得靠定时任务——一段按时间自己跑起来的代码。

    02

    最简单的定时任务：一个循环

    不用装任何库。asyncio 本身就够：

    tasks.py
    import asyncioimport loggingfrom aiogram import Botlog = logging.getLogger(__name__)async def expire_loop(bot: Bot, interval: int = 300) -> None:    """每 5 分钟扫一次到期会员。"""    while True:        try:            n = await handle_expired(bot)            if n:                log.info("处理了 %d 个到期会员", n)        except Exception:            # 这个 except 不能省 —— 少了它，一次异常整个循环就死了，            # 而且死得悄无声息，你要过几天才发现没人被踢            log.exception("扫描出错")        await asyncio.sleep(interval)

    挂到启动流程里：

    bot.py
    async def main() -> None:    ...    task = asyncio.create_task(tasks.expire_loop(bot))    try:        await dp.start_polling(bot)    finally:        task.cancel()          # 退出时把它收掉
    ⬤ 循环里的 try 是必须的

    没有它，任何一次异常都会让整个循环退出 —— 而 asyncio.create_task 创建的任务出错是不会打印任何东西的，除非你去 await 它。
    结果就是：bot 看起来一切正常，但定时任务几天前就死了，没人告诉你。

    03

    那个真正的坑：他刚好在这一刻续费

    扫描的逻辑很自然：查出所有到期的人 → 逐个踢出去。

    但「查出来」和「踢出去」之间隔着时间——你得挨个调 Telegram API，每次几百毫秒，一百个人就是几十秒。在这几十秒里，某个人付款续费了。

    我实测了这个场景：

    扫描写法续费成功被踢
    查出来直接踢是是付了钱还被踢出群。这是能上社交媒体的那种事故
    踢之前再确认一次是否正常
    认出来了吗——这就是第 6 章那个形状：先查一下，再根据查到的结果去做。只不过这次两个动作之间隔了几十秒，缝大得多。

    解法：动手的那一刻再确认一次

    storage.py
    def mark_kicked(user_id: int, now: int) -> bool:    """标记为已踢。如果这期间他续费了，返回 False。"""    with connect() as conn:        cur = conn.execute(            "UPDATE members SET kicked = 1 "            "WHERE user_id = ? AND expire_at < ? AND kicked = 0",            (user_id, now),                # ← 到期条件再查一遍        )        return cur.rowcount > 0

    扫描主体变成这样：

    tasks.py
    async def handle_expired(bot: Bot) -> int:    now = int(time.time())    done = 0    for user_id in storage.list_expired(now):        # 先在数据库里抢占，抢不到说明他续费了        if not storage.mark_kicked(user_id, now):            log.info("跳过 %s：期间续费了", user_id)            continue        try:            await bot.ban_chat_member(GROUP_ID, user_id)            await bot.unban_chat_member(GROUP_ID, user_id)   # 踢出但允许再进            done += 1        except TelegramBadRequest:            # 人早就不在群里了，标记保持，不用管            pass        await asyncio.sleep(0.5)      # 别把 API 打爆（第 7 章）    return done
    ✓ 顺序很重要：先改数据库，再调 API

    反过来的话——先踢人再标记——中间要是崩了，人被踢了但数据库不知道，下次扫描会再踢一遍。
    而按这个顺序，最坏情况是「标记了但没踢成」，用户还在群里多待一会儿。两种错都会发生，选那个后果轻的。

    顺带说一个 Telegram 的怪癖：ban_chat_member 是永久封禁，被封的人以后想续费都进不来。所以要紧接着 unban_chat_member —— 这一对连用的效果才是「踢出去，但允许再回来」。

    04

    提醒类任务：别一天发八遍

    踢人是一次性的，提醒不是。「到期前三天提醒」这种任务，如果你每 5 分钟扫一次，用户三天里会收到 864 条提醒。

    解法还是记账 —— 跟第 6 章的幂等键一个思路：

    storage.py
    CREATE TABLE IF NOT EXISTS notified (    user_id INTEGER NOT NULL,    kind    TEXT    NOT NULL,      # 'expire_3d' / 'expire_1d'    PRIMARY KEY (user_id, kind))def notify_once(user_id: int, kind: str) -> bool:    """这类提醒还没发过就返回 True。"""    with connect() as conn:        try:            conn.execute(                "INSERT INTO notified (user_id, kind) VALUES (?, ?)",                (user_id, kind),            )            return True        except sqlite3.IntegrityError:            return False      # 发过了

    续费成功时把这个用户的记录删掉，下个周期才能重新提醒。

    05

    「每天早上八点」怎么写

    固定间隔的循环做不了这个。两种办法：

    办法一：自己算下次触发时间

    tasks.py · 不用装东西
    from datetime import datetime, timedeltafrom zoneinfo import ZoneInfoTZ = ZoneInfo("Asia/Shanghai")async def daily_at(hour: int, coro_func, *args) -> None:    """每天在指定小时跑一次。"""    while True:        now = datetime.now(TZ)        nxt = now.replace(hour=hour, minute=0, second=0, microsecond=0)        if nxt <= now:            nxt += timedelta(days=1)      # 今天这个点过了，等明天        await asyncio.sleep((nxt - now).total_seconds())        try:            await coro_func(*args)        except Exception:            log.exception("每日任务出错")

    注意用了带时区的 datetime.now(TZ)——第 5 章的教训，服务器时区跟你想的不是一回事。

    办法二：用 APScheduler

    要是定时任务超过三四个，或者需要 cron 表达式，就该上库了：

    pip install apscheduler
    from apscheduler.schedulers.asyncio import AsyncIOSchedulersched = AsyncIOScheduler(timezone="Asia/Shanghai")sched.add_job(handle_expired, "interval", minutes=5, args=[bot])sched.add_job(daily_rank, "cron", hour=8, args=[bot])sched.start()

    一两个任务用手写循环就行，别急着加依赖。

    06

    三个容易忽略的细节

    一、启动时不要立刻扫

    如果你在 while True 开头就执行任务，那么每次重启都会触发一轮扫描。改代码调试时反复重启，就会反复扫。先 sleep 再干活，或者加个启动延迟。

    二、任务时间不能超过间隔

    每 5 分钟扫一次，但某次扫描花了 8 分钟怎么办？用上面那种「干完再 sleep」的写法不会重叠——下一轮会自然延后。但如果你用 APScheduler，默认是会重叠的，要加 max_instances=1。

    三、时间戳统一用 UTC 存

    数据库里存 expire_at 这种时间点，一律存 Unix 时间戳（int(time.time())）。它没有时区概念，比较大小永远正确。只在显示给用户的时候才转成他的时区。

    ◆ 存字符串日期是个陷阱

    存 "2026-08-21 10:00" 这种字符串，你根本不知道它是哪个时区的。半年后你自己也说不清 —— 而这时候数据已经有几万条了。
    时间点存时间戳，日期（比如签到的「哪一天」）才存字符串。

    07

    验一遍

    怎么试应该看到
    把 interval 临时改成 5 秒，手动把某人的 expire_at 改成过去几秒内他被处理，日志里有记录
    在扫描的循环里加 time.sleep(3)，扫描期间手动改回 expire_at日志出现「跳过：期间续费了」，人没被踢
    在 handle_expired 里故意 1/0日志有堆栈，下一轮照常执行 —— 循环没死
    连续重启 bot 三次不会重复发提醒（notified 表挡住了）
    Ctrl+C 退出没有「Task was destroyed but it is pending」的警告 —— 说明 cancel 生效了
    第二条最重要。它是这一章唯一真正难的地方，也是最容易上线之后才发现的。

    这一章的要点

    1. 定时任务就是一个 while True + sleep，但循环里必须包 try，否则一次异常就悄悄死掉
    2. 「查出来」和「动手做」之间会变天 —— 动手那一刻要把条件再确认一次（又是第 6 章那个形状）
    3. 先改数据库再调 API。两种错都会发生，选后果轻的那个
    4. 踢人要 ban + unban 连用，否则对方永远进不来
    5. 提醒类任务要记账，否则一天发几百遍
    6. 时间点存 Unix 时间戳，别存字符串
    ✓ 试着改一个

    给签到 bot 加个「三天没签到就提醒一次」：每天早上跑一次，找出 last_checkin 距今超过三天、且这个周期还没提醒过的人，私聊一句。
    难点在于「这个周期」怎么定义 —— 他签到之后要能重新提醒，所以签到时要清掉 notified 里对应的记录。

    定时任务跑起来了吗？

    定时任务的问题往往过几天才暴露。你遇到什么情况，或者有别的定时需求不知道怎么写，都可以说。留不留联系方式都行。