如何存储python-telegram-bot的ConversationHandler状态解决Heroku重启后需重发命令问题
将ConversationHandler状态存储到数据库自动恢复的方案完全可行,是该场景下的最优解,具体实现方案如下:
方案1:基于Firebase实现会话状态持久化
python-telegram-bot框架原生提供了BasePersistence抽象类,用于对接外部存储实现会话、用户、聊天数据的持久化,你已经有Firebase账号,直接实现适配类即可:
- 第一步:安装依赖,需要
python-telegram-bot、firebase-admin两个库 - 第二步:实现Firebase持久化类,参考代码如下:
from typing import Dict, Tuple, Any, Optional from telegram.ext import BasePersistence import firebase_admin from firebase_admin import credentials, firestore class FirebasePersistence(BasePersistence): def __init__(self, firebase_cred_path: str): # 配置是否存储用户、聊天、机器人全局数据 super().__init__(store_user_data=True, store_chat_data=True, store_bot_data=True) # 初始化Firebase连接 cred = credentials.Certificate(firebase_cred_path) firebase_admin.initialize_app(cred) self.db = firestore.client() self.conv_collection = self.db.collection("conversation_states") self.user_data_collection = self.db.collection("user_data") self.chat_data_collection = self.db.collection("chat_data") def get_conversations(self, name: str) -> Dict[Tuple[int, int], Optional[Any]]: # 启动时拉取对应会话的所有状态 doc = self.conv_collection.document(name).get() raw_data = doc.to_dict() if doc.exists else {} # 将Firebase存储的字符串key转回(chat_id, user_id)元组 return {tuple(map(int, k.split("_"))): v for k, v in raw_data.items()} def update_conversation(self, name: str, key: Tuple[int, int], new_state: Optional[Any]) -> None: # 会话状态变更时写入Firebase str_key = f"{key[0]}_{key[1]}" self.conv_collection.document(name).set({str_key: new_state}, merge=True) # 以下为用户数据、聊天数据持久化的实现,不需要可以直接返回空字典 def get_user_data(self) -> Dict[int, Any]: doc = self.user_data_collection.document("all").get() return {int(k): v for k, v in doc.to_dict().items()} if doc.exists else {} def update_user_data(self, user_id: int, data: Any) -> None: self.user_data_collection.document("all").set({str(user_id): data}, merge=True) def get_chat_data(self) -> Dict[int, Any]: doc = self.chat_data_collection.document("all").get() return {int(k): v for k, v in doc.to_dict().items()} if doc.exists else {} def update_chat_data(self, chat_id: int, data: Any) -> None: self.chat_data_collection.document("all").set({str(chat_id): data}, merge=True) def get_bot_data(self) -> Dict: return {} def update_bot_data(self, data: Dict) -> None: pass
- 第三步:初始化机器人时传入持久化实例即可
from telegram.ext import ApplicationBuilder, CommandHandler, ConversationHandler, MessageHandler, filters # 初始化持久化实例,填入你的Firebase凭证文件路径 persistence = FirebasePersistence("your_firebase_cred.json") # 构建Application时传入persistence参数 application = ApplicationBuilder().token("YOUR_BOT_TOKEN").persistence(persistence).build() # 后续正常注册ConversationHandler即可,不需要修改原有业务逻辑 conv_handler = ConversationHandler( entry_points=[CommandHandler("start", start)], states={...}, fallbacks=[...] ) application.add_handler(conv_handler)
完成上述配置后,每次会话状态变更都会自动同步到Firebase,脚本重启时框架会自动拉取所有保存的会话状态恢复,完全不需要用户重新发送触发命令。
方案2:更换无自动重启的托管平台
如果不需要保留Heroku部署,也可以选择支持长期驻留进程的托管服务,不会定期自动重启脚本:
- 云服务器类:阿里云ECS、腾讯云轻量应用服务器、DigitalOcean Droplet、Vultr VPS,自行部署Python环境后运行脚本即可长期稳定运行
- 常驻Serverless类:Google Cloud Run 配置常驻实例、AWS Lambda配置预发并发,也可以避免冷启动自动重启
内容的提问来源于stack exchange,提问作者Rahasyar Desh
相关产品推荐
相关产品推荐

