python-telegram-bot重启后如何让用户恢复之前的会话状态？

核心思路 #

把用户的会话状态（比如当前所处交互步骤、临时输入数据等）持久化到外部存储（数据库、文件等），机器人重启后从存储中读取状态，恢复用户的交互上下文，无需重新发送/start指令。

具体实现方案 #

1. 选择持久化存储方式 #

• 文件存储：适合小型机器人，用JSON、Pickle等格式保存状态。比如按用户ID创建单独JSON文件，或在统一JSON文件中用用户ID作为键区分状态。
• 数据库存储：适合中大型机器人，优先选SQLite（轻量无需额外服务）、Redis（高频读写场景），也可使用MySQL、PostgreSQL等。

2. 结合python-telegram-bot的会话管理实现 #

以python-telegram-bot v13+的ConversationHandler为例（v20+逻辑一致，仅接口细节略有调整）：

示例：用SQLite存储会话状态

首先初始化数据库表：

import sqlite3

def init_db():
    conn = sqlite3.connect('bot_states.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS user_states
                 (user_id INTEGER PRIMARY KEY, chat_id INTEGER, state TEXT, data TEXT)''')
    conn.commit()
    conn.close()

实现状态的保存与读取函数：

def save_user_state(user_id, chat_id, state, data=None):
    conn = sqlite3.connect('bot_states.db')
    c = conn.cursor()
    # 用REPLACE实现状态的更新或插入
    c.execute('''REPLACE INTO user_states (user_id, chat_id, state, data)
                 VALUES (?, ?, ?, ?)''', (user_id, chat_id, state, str(data) if data else None))
    conn.commit()
    conn.close()

def load_user_state(user_id):
    conn = sqlite3.connect('bot_states.db')
    c = conn.cursor()
    c.execute('SELECT state, data, chat_id FROM user_states WHERE user_id = ?', (user_id,))
    result = c.fetchone()
    conn.close()
    if result:
        state, data_str, chat_id = result
        # 将存储的字符串格式数据转回字典
        try:
            data = eval(data_str) if data_str else {}
        except:
            data = {}
        return state, data, chat_id
    return None, None, None

自定义持久化类对接ConversationHandler：

from telegram.ext import BasePersistence, PersistenceInput

class CustomPersistence(BasePersistence):
    def __init__(self):
        super().__init__(persistence_input=PersistenceInput(chat_data=False, user_data=True, bot_data=False))
    
    def get_user_data(self):
        # 从数据库加载所有用户的会话数据
        conn = sqlite3.connect('bot_states.db')
        c = conn.cursor()
        c.execute('SELECT user_id, data FROM user_states')
        user_data = {}
        for row in c.fetchall():
            user_id, data_str = row
            try:
                data = eval(data_str) if data_str else {}
            except:
                data = {}
            user_data[user_id] = data
        conn.close()
        return user_data
    
    def update_user_data(self, user_id, data):
        # 更新用户数据到数据库
        save_user_state(user_id, None, None, str(data))
    
    def get_conversations(self, name):
        # 加载所有用户的对话状态
        conn = sqlite3.connect('bot_states.db')
        c = conn.cursor()
        c.execute('SELECT user_id, chat_id, state FROM user_states')
        conversations = {}
        for row in c.fetchall():
            user_id, chat_id, state = row
            conversations[(user_id, chat_id)] = state
        conn.close()
        return conversations
    
    def update_conversation(self, name, key, new_state):
        # 更新对话状态到数据库
        user_id, chat_id = key
        save_user_state(user_id, chat_id, new_state)
    
    # 以下方法按需实现，此处留空或返回默认值
    def get_chat_data(self):
        return {}
    def update_chat_data(self, chat_id, data):
        pass
    def get_bot_data(self):
        return {}
    def update_bot_data(self, data):
        pass

创建Updater时传入自定义持久化类：

from telegram.ext import Updater

updater = Updater("你的机器人Token", persistence=CustomPersistence())
# 后续添加ConversationHandler等业务逻辑

3. 处理重启后的用户消息 #

机器人重启后，用户发送任意消息时自动读取状态并恢复：

def handle_any_message(update, context):
    user_id = update.effective_user.id
    state, data, chat_id = load_user_state(user_id)
    if state:
        # 恢复用户的会话数据
        context.user_data.update(data)
        # 直接跳转到之前的状态处理逻辑
        return state
    else:
        update.message.reply_text("请发送/start开始交互")
        return ConversationHandler.END

注意事项 #

• 存储的状态数据只保留必要信息，避免冗余占用存储空间。
• 若使用Pickle存储数据，需注意安全风险，避免加载恶意数据。
• python-telegram-bot v20+版本的Application持久化接口略有不同，可参考官方文档调整代码。

内容的提问来源于stack exchange，提问作者Arda Senoglu