pyTelegramBotAPI多会话数据隔离问题解决方案咨询

你目前用全局列表pos_array、pos_count和全局变量j存储用户数据，这些变量是所有用户共享的，不同用户的操作会互相覆盖，自然导致数据串用的问题。

不需要直接上数据库，先从轻量方案入手，再根据需求考虑持久化选项：

1. 自定义用户会话类（轻量场景首选） #

创建一个类封装每个用户的专属数据，用字典关联用户ID和对应的类实例，实现数据完全隔离。

class UserSession:
    def __init__(self):
        self.pos_array = []
        self.pos_count = []
        self.j = 0

# 存储所有用户的会话实例，key为用户ID
user_sessions = {}

# 获取或创建用户的专属会话
def get_user_session(user_id):
    if user_id not in user_sessions:
        user_sessions[user_id] = UserSession()
    return user_sessions[user_id]

修改原有函数，替换全局变量为会话实例的属性：

@bot.callback_query_handler(func=lambda call: call.data == "newpositions")
def set_positions(call):
    user_id = call.from_user.id
    session = get_user_session(user_id)
    if len(session.pos_array) > 1:
        bot.send_message(user_id, "Positions already set")
    else: 
        print(call)
        pos_num_reply = bot.send_message(user_id, "How many positions do you have?")
        bot.register_next_step_handler(pos_num_reply, call_positions)

def call_positions(message):
    user_id = message.from_user.id
    session = get_user_session(user_id)
    session.pos_count.append(int(float(message.text)))
    print("Pos_count array->", session.pos_count)
    if session.j < session.pos_count[0]:
        set_pos = bot.send_message(user_id, f"Set Position {session.j+1}")
        bot.register_next_step_handler(set_pos, set_position)
    else:
        bot.send_message(user_id, f"Your positions:\n{chr(92).join(map(str, session.pos_array))}")

其他函数同理，把所有全局变量的引用替换为session.xxx即可。

2. 用嵌套字典存储（更简洁的轻量方案） #

如果不想写类，直接用嵌套字典存储用户数据：

user_data = {}

def get_user_data(user_id):
    if user_id not in user_data:
        user_data[user_id] = {
            "pos_array": [],
            "pos_count": [],
            "j": 0
        }
    return user_data[user_id]

在函数中调用user_data = get_user_data(user_id)，通过user_data["pos_array"]访问对应数据，效果和类方案一致。

3. 接入数据库（持久化需求） #

如果需要Bot重启后数据不丢失，或者数据量较大，就需要数据库：

• SQLite：轻量文件型数据库，Python自带sqlite3库，适合小型Bot。
• Redis：内存型数据库，速度快，适合会话数据存储，支持持久化。
• MySQL/PostgreSQL：适合大型Bot，支持复杂查询。

以SQLite为例，先创建用户表：

import sqlite3

conn = sqlite3.connect('bot_users.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS user_positions
             (user_id INTEGER PRIMARY KEY, pos_array TEXT, pos_count TEXT, j INTEGER)''')
conn.commit()

封装读写函数：

def get_user_data_from_db(user_id):
    c.execute('SELECT pos_array, pos_count, j FROM user_positions WHERE user_id = ?', (user_id,))
    result = c.fetchone()
    if result:
        pos_array = [float(x) for x in result[0].split(',')] if result[0] else []
        pos_count = [int(x) for x in result[1].split(',')] if result[1] else []
        j = result[2]
        return {"pos_array": pos_array, "pos_count": pos_count, "j": j}
    else:
        c.execute('INSERT INTO user_positions VALUES (?, ?, ?, ?)', (user_id, "", "", 0))
        conn.commit()
        return {"pos_array": [], "pos_count": [], "j": 0}

def update_user_data(user_id, pos_array, pos_count, j):
    pos_array_str = ','.join(map(str, pos_array))
    pos_count_str = ','.join(map(str, pos_count))
    c.execute('UPDATE user_positions SET pos_array = ?, pos_count = ?, j = ? WHERE user_id = ?',
              (pos_array_str, pos_count_str, j, user_id))
    conn.commit()

之后在函数中调用这些数据库操作函数即可。

关于pyTelegramBotAPI的会话支持 #

pyTelegramBotAPI本身没有内置多会话管理类，用户数据隔离需要自己实现，上面的自定义类/字典方案是最常用的方式。

• 清理过期会话：定时删除长时间未活跃用户的会话数据，避免内存占用过高。
• 异常处理：添加用户输入非数字等异常情况的捕获，防止Bot崩溃。
• 避免全局变量：除了用户会话存储，尽量不使用全局变量，所有用户相关数据都绑定到用户ID。

内容的提问来源于stack exchange，提问作者Yuriy Timoshenkov