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

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

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

在使用 python-telegram-bot（v20+）构建定时告警 Bot 时，一个典型需求是：Bot 重启后，自动检查数据库中已过期但尚未发送的告警任务，并立即向对应用户推送提醒（例如“服务已恢复，您有一条延迟告警”）。然而，直接在 main() 函数同步上下文中调用 application.bot.send_message(…) 会触发 RuntimeWarning: coroutine was never awaited；若强行用 asyncio.run() 包裹，则可能因事件循环已被 Application 启动并关闭而导致 RuntimeError: Event loop is closed。

根本原因在于：Application 的 bot 属性所有 API 方法（如 send_message）均为 协程函数（coroutine），必须在运行中的事件循环内 await 调用，而 main() 是同步入口，且 application.run_polling() 会接管并管理专属事件循环。

✅ 正确做法是：将重启后的通知逻辑封装为异步任务，并通过 application.bot 的 initialize() 和 shutdown() 生命周期钩子，或更推荐的方式——在 run_polling() 启动前，使用 application.create_task() 提交到 Bot 的主事件循环中执行。

以下是经过验证的完整实现方案：

✅ 推荐方案：使用 application.create_task() 在主事件循环中安全发送

import asyncio from datetime import datetime import sqlite3 from telegram.ext import Application, CommandHandler, MessageHandler, filters, ApplicationBuilder from telegram import Update  # 假设已初始化数据库连接 conn = sqlite3.connect("alarms.db") cursor = conn.cursor()  def main() -> None:     application = ApplicationBuilder().token("YOUR_TOKEN").build()      # 注册 handlers（略）     # application.add_handler(...)      # --- 关键：Bot 启动后立即执行的异步初始化逻辑 ---     async def on_startup():         print("✅ Bot started. Loading pending alarms...")         current_time = datetime.now()          cursor.execute("SELECT id, chat_id, alarm_time, message FROM alarms")         rows = cursor.fetchall()          for row in rows:             alarm_id, chat_id, alarm_time_str, msg = row             try:                 alarm_time = datetime.strptime(alarm_time_str, '%Y-%m-%d %H:%M:%S')             except ValueError:                 continue              if alarm_time < current_time:                 # 过期告警：立即发送 + 清理 DB                 try:                     await application.bot.send_message(                         chat_id=chat_id,                         text=f"⏰ 延迟告警：{msg}n（原定于 {alarm_time_str}）"                     )                     cursor.execute("DELETE FROM alarms WHERE id = ?", (alarm_id,))                     conn.commit()                     print(f"✅ Sent & deleted alarm #{alarm_id} for chat {chat_id}")                 except Exception as e:                     print(f"❌ Failed to send alarm #{alarm_id}: {e}")      # 在 run_polling() 之前，将 on_startup 提交至 Bot 的事件循环     application.create_task(on_startup())      # 启动 Bot（会自动等待 on_startup 完成）     application.run_polling(allowed_updates=Update.ALL_TYPES)  if __name__ == "__main__":     main()

⚠️ 注意事项与最佳实践

• 不要在 main() 中直接 await：main() 是同步函数，不能 await；必须通过 application.create_task() 或 application.bot.initialize() 钩子提交协程。
• 避免 asyncio.run()：application.run_polling() 已启动专属事件循环，外部再调用 asyncio.run() 会创建新循环，导致冲突和 Event loop is closed 错误。
• DB 操作需线程安全：SQLite 默认不支持多线程并发写入。若 Bot 后续引入并发 Job（如多个 run_once），建议：
  • 使用 threading.Lock 包裹 cursor.execute() / conn.commit()；
  • 或改用支持异步的数据库驱动（如 aiosqlite）。
• 错误处理不可省略：网络波动或用户退订可能导致 send_message 抛出 Forbidden, BadRequest 等异常，务必用 try/except 捕获并记录，避免中断整个初始化流程。
• 考虑用户体验：重启通知应简洁友好，避免高频打扰。可增加开关（如 /notify_on_restart on/off）由用户控制。

✅ 总结

Bot 重启后发送消息的核心原则是：一切 Bot API 调用必须在 Application 托管的事件循环中 await 执行。application.create_task() 是最轻量、最可靠的入口方式，它确保你的异步初始化逻辑与 Bot 生命周期完全对齐。配合健壮的异常处理与数据库操作规范，即可实现稳定、可维护的告警恢复机制。