本文详解如何在 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 实例可用、网络连接建立）之后执行。
以下是推荐实现方案：
✅ 正确实现：使用 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 创建连接，或改用线程安全的连接池。
✅ 总结
| 错误方式 | 正确方式 |
|---|---|
| application.bot.send_message(…)（未 await）→ RuntimeWarning | await application.bot.send_message(…)（在 post_init 中） |
| asyncio.run(…) 在 main() 中 → Event loop is closed | 使用 application.post_init 钩子，交由 Bot 主循环调度 |
| 启动时硬编码 current_time → 过期判断不准 | 在 post_init 内动态获取 datetime.now() |
通过 post_init，你不仅能可靠发送重启通知，还可扩展用于加载缓存、预热数据、健康检查等初始化任务——这是 python-telegram-bot v20+ 推荐的标准模式。