如何在 Telegram Bot 重启后自动发送通知消息

本文详解如何在 Python-telegram-bot v20+ 中，于 Bot 启动完成、应用初始化就绪后，安全、异步地向用户发送重启通知或过期告警消息，避免 RuntimeWarning: coroutine was never awaited 和 Event loop is closed 等常见异步错误。

本文详解如何在 python-telegram-bot v20+ 中，于 bot 启动完成、应用初始化就绪后，安全、异步地向用户发送重启通知或过期告警消息，避免 `runtimewarning: coroutine was never awaited` 和 `event loop is closed` 等常见异步错误。

在使用 python-telegram-bot（v20+）构建定时告警 Bot 时，一个典型需求是：Bot 重启后，自动检查数据库中已过期但尚未触发的 alarm 记录，并立即向对应用户发送提醒（例如“您的定时任务已过期，请重新设置”）。然而，直接在 main() 函数同步上下文中调用 application.bot.send_message(...) 会引发 RuntimeWarning: coroutine 'ExtBot.send_message' was never awaited —— 因为该方法是协程（coroutine），必须 await 才能执行。

更关键的是，若尝试用 asyncio.run() 在非事件循环环境中手动启动新循环（如 asyncio.run(send_notification(...))），则会在后续 application.run_polling() 启动主事件循环时导致 RuntimeError: Event loop is closed，因为 asyncio.run() 会创建并关闭独立循环，与 Bot 框架管理的主循环冲突。

✅ 正确做法是：将消息发送逻辑注册为 Bot 启动后的异步回调（post-startup hook），确保它运行在 Bot 的主事件循环中，且在 Application 完全就绪（包括 bot 实例可用、网络连接建立）之后执行。

Lenso.ai

AI反向图像搜索

下载

以下是推荐实现方案：

✅ 正确实现：使用 application.post_init 钩子

import asyncio
from datetime import datetime
import sqlite3
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ApplicationBuilder
from telegram import Update

# 假设已定义 start/set_timer/... 等 handler
# 数据库连接（注意：实际项目中建议封装为全局或依赖注入）
conn = sqlite3.connect("alarms.db")
cursor = conn.cursor()

async def post_init(application: Application) -> None:
    """Bot 启动完成后执行的异步初始化钩子"""
    print("✅ Bot 已启动，正在恢复过期告警...")

    # 查询所有已过期的 alarm（注意：需确保 current_time 是动态获取的）
    current_time = datetime.now()
    cursor.execute("SELECT id, chat_id, scheduled_time, message FROM alarms WHERE scheduled_time < ?", (current_time.strftime('%Y-%m-%d %H:%M:%S'),))
    expired_alarms = cursor.fetchall()

    for alarm_id, chat_id, scheduled_time, message in expired_alarms:
        try:
            # ✅ 安全发送：await 在主事件循环中执行
            await application.bot.send_message(
                chat_id=chat_id,
                text=f"⏰ 提醒：您设置于 {scheduled_time} 的告警已过期。\n内容：{message}\n（Bot 重启后自动通知）"
            )
            # 可选：清理已处理记录
            cursor.execute("DELETE FROM alarms WHERE id = ?", (alarm_id,))
            conn.commit()
            print(f"✅ 已通知用户 {chat_id}（alarm #{alarm_id}）")
        except Exception as e:
            print(f"❌ 发送失败（用户 {chat_id}）：{e}")

def main() -> None:
    application = ApplicationBuilder().token("YOUR_TOKEN").build()

    # 注册命令处理器（略去具体 handler 定义）
    application.add_handler(CommandHandler(["start", "help"], start))
    application.add_handler(CommandHandler("set", set_timer))
    application.add_handler(CommandHandler("unset", unset))
    application.add_handler(CommandHandler("s", set_alarm))
    application.add_handler(MessageHandler(filters.TEXT & (~filters.COMMAND), def_reply))
    application.add_handler(MessageHandler(filters.COMMAND, unknown))

    # ? 关键：注册 post_init 钩子（必须在 run_polling 前设置）
    application.post_init = post_init

    # 启动 Bot（自动运行主事件循环）
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

⚠️ 注意事项与最佳实践

• 不要在 main() 同步函数体中 await 或 asyncio.run()：main() 是同步入口，而 Application 的生命周期由其内部事件循环管理。
• post_init 是唯一安全时机：它在 Application.initialize() 完成、bot 实例已认证并可通信后被调用，且以 await 方式执行，天然兼容协程。
• 时间比较务必使用同一时区：示例中 scheduled_time 存为字符串，建议数据库字段改为 TEXT 并统一存为 ISO 格式（如 '2024-02-15T13:56:00'），Python 中用 datetime.fromisoformat() 解析，避免因格式/时区导致误判。
• 异常处理不可省略：用户可能已退群或封禁 Bot，send_message 可能抛出 Forbidden, BadRequest 等异常，应捕获并记录，避免中断整个 post_init 流程。
• 数据库连接线程安全：SQLite 默认不支持多线程并发写入。若 Bot 启动后还需在 Job 或 Handler 中操作数据库，建议使用 check_same_thread=False 创建连接，或改用线程安全的连接池。

✅ 总结

通过 post_init，你不仅能可靠发送重启通知，还可扩展用于加载缓存、预热数据、健康检查等初始化任务——这是 python-telegram-bot v20+ 推荐的标准模式。