Python 电报机器人。如何将 ConversationQueryCallback 与实例方法一起用作回调?
Python Telegram Bot. How to use ConversationQueryCallback with instance method as callback?
我正在围绕库 python-telegram-bot 设置一个包装器,只是因为我不喜欢构建一个编写 100 万行备用函数和调度程序添加的机器人。
class TelegramBot:
_types = {
"_command_callback": CommandHandler,
"_message_callback": MessageHandler,
"_conversation_callback": ConversationHandler
}
def __init__(self, token=token):
self.updater = Updater(token, use_context=True)
self.dispatcher = self.updater.dispatcher
self.add_dispatchers()
def add_dispatchers(self):
for _type, _class in self._types.items():
handler_functions = list(filter(lambda x: x.endswith(_type), dir(self)))
for handler in handler_functions:
attr_name = handler.replace('callback', "handler")
if _type == '_message_callback':
setattr(self, attr_name, _class(Filters.text, getattr(self, handler)))
elif _type == '_command_callback':
setattr(self, attr_name, _class(handler.replace(_type, ''), getattr(self, handler)))
elif _type == '_conversation_callback':
setattr(self, attr_name, _class(*getattr(self, handler)))
self.dispatcher.add_handler(getattr(self, attr_name))
def get_handler(self, name, type):
_name = f"{name}_{type}_handler"
return getattr(self, _name)
def start_polling(self):
self.updater.start_polling()
self.updater.idle()
有了这个简单的 class,现在我可以简单地做一些事情:
class MyCustomBot(TelegramBot):
def test_command_callback(self, update, context):
update.message.reply_text("Testing command")
def echo_message_callback(self, update, context):
update.message.reply_text(update.message.text)
my_bot = MyCustomBot()
my_bot.start_polling()
在上面的这 7 行中,我定义了一个命令处理程序和一个简单的消息处理程序。挺干净的。
现在我想添加一个对话处理程序。在 class 之外,一切正常,但如果我这样定义 MyCustomBot:
class MyCustomBot(TelegramBot):
def test_command_callback(self, update, context):
update.message.reply_text("Testing command")
def echo_message_callback(self, update, context):
update.message.reply_text(update.message.text)
@property
def test_conversation_callback(self):
entry_points = [self.get_handler('search', 'command')]
states = {
1: [CallbackQueryHandler(self.test_command_callback)]
}
fallbacks = entry_points
return entry_points, states, fallbacks
def search_command_callback(self, update, context):
keyboard = [
[InlineKeyboardButton("1", callback_data=str(1)),
InlineKeyboardButton("2", callback_data=str(2))]
]
reply_markup = InlineKeyboardMarkup(keyboard)
# Send message with text and appended InlineKeyboard
update.message.reply_text(
"Start handler, Choose a route",
reply_markup=reply_markup
)
return 1
bot = MyCustomBot()
bot.start_polling()
没用。如果我触发 /search
,它会正确地向我发送键盘,但是如果我按下一个按钮,我会看到一条警告通过日志显示 "conversation returned status None".
我想问题可能出在这里?
states = {
1: [CallbackQueryHandler(self.test_command_callback)]
}
self.test_command_callback
是一个实例方法,CallbackQueryHandler可以用它作为回调吗?
或者我还遗漏了什么?
终于发现自己哪里做错了
我将 search_command_callback
注册为纯 CommandHandler
并且 也注册到 ConversationHandler
的 entry_points
。
这不起作用。入口点不能也注册 CommandHandlers。
我不知道这是否记录在案。在几个小时的调试中,我没有发现文档中报告了这一点。
我正在围绕库 python-telegram-bot 设置一个包装器,只是因为我不喜欢构建一个编写 100 万行备用函数和调度程序添加的机器人。
class TelegramBot:
_types = {
"_command_callback": CommandHandler,
"_message_callback": MessageHandler,
"_conversation_callback": ConversationHandler
}
def __init__(self, token=token):
self.updater = Updater(token, use_context=True)
self.dispatcher = self.updater.dispatcher
self.add_dispatchers()
def add_dispatchers(self):
for _type, _class in self._types.items():
handler_functions = list(filter(lambda x: x.endswith(_type), dir(self)))
for handler in handler_functions:
attr_name = handler.replace('callback', "handler")
if _type == '_message_callback':
setattr(self, attr_name, _class(Filters.text, getattr(self, handler)))
elif _type == '_command_callback':
setattr(self, attr_name, _class(handler.replace(_type, ''), getattr(self, handler)))
elif _type == '_conversation_callback':
setattr(self, attr_name, _class(*getattr(self, handler)))
self.dispatcher.add_handler(getattr(self, attr_name))
def get_handler(self, name, type):
_name = f"{name}_{type}_handler"
return getattr(self, _name)
def start_polling(self):
self.updater.start_polling()
self.updater.idle()
有了这个简单的 class,现在我可以简单地做一些事情:
class MyCustomBot(TelegramBot):
def test_command_callback(self, update, context):
update.message.reply_text("Testing command")
def echo_message_callback(self, update, context):
update.message.reply_text(update.message.text)
my_bot = MyCustomBot()
my_bot.start_polling()
在上面的这 7 行中,我定义了一个命令处理程序和一个简单的消息处理程序。挺干净的。
现在我想添加一个对话处理程序。在 class 之外,一切正常,但如果我这样定义 MyCustomBot:
class MyCustomBot(TelegramBot):
def test_command_callback(self, update, context):
update.message.reply_text("Testing command")
def echo_message_callback(self, update, context):
update.message.reply_text(update.message.text)
@property
def test_conversation_callback(self):
entry_points = [self.get_handler('search', 'command')]
states = {
1: [CallbackQueryHandler(self.test_command_callback)]
}
fallbacks = entry_points
return entry_points, states, fallbacks
def search_command_callback(self, update, context):
keyboard = [
[InlineKeyboardButton("1", callback_data=str(1)),
InlineKeyboardButton("2", callback_data=str(2))]
]
reply_markup = InlineKeyboardMarkup(keyboard)
# Send message with text and appended InlineKeyboard
update.message.reply_text(
"Start handler, Choose a route",
reply_markup=reply_markup
)
return 1
bot = MyCustomBot()
bot.start_polling()
没用。如果我触发 /search
,它会正确地向我发送键盘,但是如果我按下一个按钮,我会看到一条警告通过日志显示 "conversation returned status None".
我想问题可能出在这里?
states = {
1: [CallbackQueryHandler(self.test_command_callback)]
}
self.test_command_callback
是一个实例方法,CallbackQueryHandler可以用它作为回调吗?
或者我还遗漏了什么?
终于发现自己哪里做错了
我将 search_command_callback
注册为纯 CommandHandler
并且 也注册到 ConversationHandler
的 entry_points
。
这不起作用。入口点不能也注册 CommandHandlers。
我不知道这是否记录在案。在几个小时的调试中,我没有发现文档中报告了这一点。