嗯，下面的代码为
/startdp = updater.dispatcher
dp.add_handler(CommandHandler('start', start))
我想添加一个处理程序，可以在一个处理程序中处理
/download_video/download_music我想到的是这样的：
dp.add_handler(CommandHandler(r'^download', download))
但是它并没有像想象的那样工作！相反，当我发送
/^download我该怎么做？
迟到的答复。一个简单的方法是解析所有短信并检查它们是否以匹配的单词开头，即：
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters
import re
def parse_msg(update, context):
if re.search("^(/test|/something_else)$", update.message.text, re.IGNORECASE | re.DOTALL):
update.message.reply_text("send your content")
def main():
updater = Updater("your_token", use_context=True)
dp = updater.dispatcher
dp.add_handler(MessageHandler(Filters.text, parse_msg)) # parses all text messages
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()
/test/something_else/setprivacy@bot_nameDisable您可以使用正则表达式创建自定义处理程序。例如，创建名为 regexp_command handler.py 的新文件。然后将下面编写的代码复制到其中。现在您可以使用 regexp 作为处理程序参数。例如，
dispatcher.add_handler(RegexpCommandHandler(r'start_[\d]+', start))import re
import warnings
from telegram import Update
from telegram.ext import Handler
class RegexpCommandHandler(Handler):
def __init__(self,
command_regexp,
callback,
filters=None,
allow_edited=False,
pass_args=False,
pass_update_queue=False,
pass_job_queue=False,
pass_user_data=False,
pass_chat_data=False):
super(RegexpCommandHandler, self).__init__(
callback,
pass_update_queue=pass_update_queue,
pass_job_queue=pass_job_queue,
pass_user_data=pass_user_data,
pass_chat_data=pass_chat_data
)
self.command_regexp = command_regexp
self.filters = filters
self.allow_edited = allow_edited
self.pass_args = pass_args
# We put this up here instead of with the rest of checking code
# in check_update since we don't wanna spam a ton
if isinstance(self.filters, list):
warnings.warn('Using a list of filters in MessageHandler is getting '
'deprecated, please use bitwise operators (& and |) '
'instead. More info: https://git.io/vPTbc.')
def check_update(self, update):
if (isinstance(update, Update)
and (update.message or update.edited_message and self.allow_edited)):
message = update.message or update.edited_message
if message.text and message.text.startswith('/') and len(message.text) > 1:
command = message.text[1:].split(None, 1)[0].split('@')
command.append(
message.bot.username) # in case the command was send without a username
print(command)
if self.filters is None:
res = True
elif isinstance(self.filters, list):
res = any(func(message) for func in self.filters)
else:
res = self.filters(message)
match = re.match(self.command_regexp, command[0])
return res and (bool(match) and command[1].lower() == message.bot.username.lower())
else:
return False
else:
return False
def handle_update(self, update, dispatcher):
optional_args = self.collect_optional_args(dispatcher, update)
message = update.message or update.edited_message
if self.pass_args:
optional_args['args'] = message.text.split()[1:]
return self.callback(dispatcher.bot, update, **optional_args)
这是一个基于@Alexandr 的答案无耻构建的更新版本。我已经简化了它，现在它可以在 v20.0 上运行：
command_regex_handler.pyimport re
from telegram import Update
from telegram.ext import Handler
class RegexpCommandHandler(Handler):
def __init__(self, command_regexp, callback, separator="_", allow_edited=True, pass_args=True):
super().__init__(callback)
self.command_regexp = command_regexp
self.separator = separator
self.allow_edited = allow_edited
self.pass_args = pass_args
def check_update(self, update):
"""
This method is called to determine if an update should be handled by this handler instance.
"""
if (isinstance(update, Update)
and (update.message or update.edited_message and self.allow_edited)):
message = update.message or update.edited_message
if message.text and message.text.startswith('/') and len(message.text) > 1:
command = message.text[1:].split(None, 1)[0].split('@')
command.append(
update.effective_user.username) # in case the command was send without a username
match = re.match(self.command_regexp, command[0])
return True and (bool(match) and command[1].lower() == update.effective_user.username.lower())
else:
return False
else:
return False
async def handle_update(self, update, application, check_result, context):
"""
This method is called if it was determined that an update should indeed be handled by this instance.
Splits the command by the defined separator and returns arguments.
"""
if self.pass_args:
message = update.message or update.edited_message
optional_args = message.text.split(self.separator)[1:]
else:
optional_args = []
return await self.callback(update, context, *optional_args)
还有一个测试机器人：
import logging
from telegram import Update
from telegram.ext import ApplicationBuilder, CallbackContext, CommandHandler
from command_regex_handler import RegexpCommandHandler
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
async def start(update: Update, context: CallbackContext.DEFAULT_TYPE):
await context.bot.send_message(chat_id=update.effective_chat.id, text="I'm a bot, please talk to me!")
async def regexyey(update: Update, context: CallbackContext.DEFAULT_TYPE, *other_args):
print('Passed arguments', other_args)
await context.bot.send_message(chat_id=update.effective_chat.id, text="worked")
if __name__ == '__main__':
application = ApplicationBuilder().token('TOKEN').build()
start_handler = CommandHandler('start', start)
application.add_handler(RegexpCommandHandler(r're->[\d]+', regexyey, separator='->'))
application.add_handler(start_handler)
application.run_polling()
据我所知，使用标准工具是不可能的。该框架的当前版本是
20.6我为我的项目编写了一个
RegexCommandHandlerimport re
from typing import Optional
from telegram.ext import BaseHandler
from telegram import Update, MessageEntity
from telegram.ext import filters as filters_module
class RegexCommandHandler(BaseHandler):
__slots__ = (
"pattern",
"filters"
)
def __init__(self, pattern: re.Pattern | str, callback, filters: Optional[filters_module.BaseFilter] = None):
super().__init__(callback)
if isinstance(pattern, str):
pattern = re.compile(pattern)
self.pattern = pattern
self.filters: filters_module.BaseFilter = (
filters if filters is not None else filters_module.UpdateType.MESSAGES
)
def check_update(self, update):
if isinstance(update, Update) and update.effective_message:
message = update.effective_message
if (
message.entities
and message.entities[0].type == MessageEntity.BOT_COMMAND
and message.entities[0].offset == 0
and message.text
and message.get_bot()
):
command = message.text[1:message.entities[0].length]
args = message.text.split()[1:]
command_parts = command.split("@")
command_parts.append(message.get_bot().username)
match = self.pattern.match(command_parts[0])
if not (match and command_parts[1].lower() == message.get_bot().username.lower()):
return None
filter_result = self.filters.check_update(update)
if filter_result:
return args, filter_result, match.groupdict()
return False
return None
def collect_additional_context(
self,
context,
update: Update,
application,
check_result,
) -> None:
if isinstance(check_result, tuple):
context.args = check_result[0]
if isinstance(check_result[1], dict):
context.update(check_result[1])
if isinstance(check_result[2], dict):
context.update(check_result[2])
它主要是常规 ol'
CommandHandlerCallbackContext看看用法：
async def handler(update, ctx):
preset_id = ctx.preset_id
# do something with preset id...
app.add_handler(
RegexCommandHandler(
re.compile(r"use_preset_(?P<preset_id>\d+)"),
callback=handler
)
)
CommandHandlerdp.add_handler(CommandHandler(('download_video', 'download_music'), download))或
dp.add_handler(CommandHandler(['download_video', 'download_music'], download))