AsyncTeleBot回调查询处理器与States结合的异常排查

核心错误原因 #

在async版本的pyTelegramBotAPI中，bot.retrieve_data()是异步函数，且返回对象并非上下文管理器（不支持with语句），直接用with调用会触发__enter__方法不存在的错误，这就是日志中ERROR - TeleBot: "__enter__"的根源。

修复后的代码 #

import asyncio
import os
from telebot.asyncio_handler_backends import State, StatesGroup
from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton
from telebot.async_telebot import AsyncTeleBot, asyncio_filters, StateMemoryStorage
from apscheduler.schedulers.asyncio import AsyncIOScheduler

#outer modules
from main_operations import calc_profit, get_attr, build_menu
from apsched import send_message_interval 

class States(StatesGroup):
    prof_nm = State()  # 调整顺序匹配流程逻辑
    prof_price = State()

async def telegram_bot():
    state_storage = StateMemoryStorage()
    bot = AsyncTeleBot(os.getenv('TG_TEST_TOKEN'), state_storage=state_storage)
    
    # 创建带商品ID的按钮菜单
    @bot.message_handler(func=lambda message: message.chat.id in users, commands=['profit'])
    async def command_profit(message):
        nm_ids = get_attr('nmId')  # 返回ID列表
        button_list = [InlineKeyboardButton(nm, callback_data=nm) for nm in nm_ids]
        reply_markup = InlineKeyboardMarkup(build_menu(button_list, n_cols=3))
        # 设置初始状态
        await bot.set_state(message.from_user.id, States.prof_nm, message.chat.id)
        await bot.send_message(chat_id=message.chat.id, text="Choose product", reply_markup=reply_markup)

    # 处理商品ID选择回调
    @bot.callback_query_handler(func=lambda callback: True, state=States.prof_nm)
    async def get_callback(callback):
        user_id = callback.from_user.id
        chat_id = callback.message.chat.id
        
        # 异步获取状态数据
        data = await bot.retrieve_data(user_id, chat_id)
        data['prof_nm'] = callback.data
        # 显式保存修改后的数据
        await bot.save_data(user_id, chat_id, data)
        
        await bot.set_state(user_id, States.prof_price, chat_id)
        await bot.send_message(chat_id, 'Write a price')
        print('retrieved!')  # 现在会正常打印

    # 处理价格输入并计算利润
    @bot.message_handler(state=States.prof_price)
    async def click_profit(message):
        user_id = message.from_user.id
        chat_id = message.chat.id
        
        data = await bot.retrieve_data(user_id, chat_id)
        nm = data['prof_nm']  # 修正变量名，与保存的键一致
        price = message.text  # text是属性，不是方法，去掉括号
        
        profit = calc_profit(nm, price) 
        await bot.send_message(chat_id, profit)
        await bot.delete_state(user_id, chat_id)
    
    # 测试定时任务功能
    @bot.message_handler(func=lambda message: message.chat.id in users, commands=['info'])
    async def get_info(message):
        scheduler = AsyncIOScheduler(timezone='Europe/Moscow')
        scheduler.add_job(send_message_interval, trigger='interval', seconds=10, kwargs={'bot': bot})
        scheduler.start()
    
    bot.add_custom_filter(asyncio_filters.StateFilter(bot))
    await bot.polling(skip_pending=True)


users = [my_chat_id]

if __name__ == "__main__":
    asyncio.run(telegram_bot())

关键修复点说明 #

1. 异步数据存取：将with bot.retrieve_data(...)替换为await bot.retrieve_data()获取数据，修改后通过await bot.save_data()显式保存，这是async版本状态管理的标准写法。
2. 变量与属性修正：click_profit中错误使用data['name']，改为data['prof_nm']匹配保存的键；message.text()改为message.text，因为text是消息对象的属性而非可调用方法。
3. 缩进修正：修复get_info函数的缩进错误，确保代码语法合法。

Telegram机器人开发通用建议 #

• 区分sync/async API：pyTelegramBotAPI的同步、异步版本API存在差异，开发时严格遵循对应模式的文档，避免混用写法。
• 状态管理规范：统一用户ID、聊天ID的获取方式（比如callback.from_user.id比callback.message.from_user.id更可靠），确保状态切换与数据保存顺序合理。
• 定时任务优化：不要在命令处理器中重复创建AsyncIOScheduler，建议全局初始化一次，避免多次启动调度器造成资源浪费。
• 错误捕获增强：在异步处理器中添加try-except块捕获异常并打印详细回溯，方便排查问题，示例：

  @bot.callback_query_handler(...)
  async def get_callback(callback):
      try:
          # 业务逻辑代码
      except Exception as e:
          print(f"回调处理出错: {e}", exc_info=True)
          await bot.send_message(callback.message.chat.id, "操作失败，请重试")
  

• 框架选型参考：如果需求涉及复杂状态管理、高并发任务，可考虑使用aiogram框架，它的设计更贴合asyncio生态，状态管理与调度器集成更流畅。

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