我们从Python开源项目中，提取了以下20个代码示例，用于说明如何使用telegram.ext.Filters.command()。
def main(): # Create the EventHandler and pass it your bot's token. updater = Updater(api_key, workers=64) # Get the dispatcher to register handlers dp = updater.dispatcher # on noncommand i.e message - return not found dp.add_handler(MessageHandler(Filters.command, maintenance)) dp.add_handler(MessageHandler(Filters.text, maintenance)) # log all errors dp.add_error_handler(error) # Start the Bot #updater.start_polling() updater.start_webhook(listen='127.0.0.1', port=int(listen_port), url_path=api_key) updater.bot.setWebhook('https://{0}/{1}'.format(domain, api_key)) # Run the bot until the you presses Ctrl-C or the process receives SIGINT, # SIGTERM or SIGABRT. This should be used most of the time, since # start_polling() is non-blocking and will stop the bot gracefully. updater.idle()
def main(): # define the updater updater = Updater(token=botconfig.bot_token) # define the dispatcher dp = updater.dispatcher # messages dp.add_handler(MessageHandler(~Filters.command, util.process_message, edited_updates=True)) dp.add_handler(CommandHandler(('start'), commands.help_command)) dp.add_handler(CommandHandler(('stats'), commands.stats_command)) dp.add_handler(CommandHandler(('globalstats'), commands.global_stats_command)) # handle errors dp.add_error_handler(error) updater.start_polling() updater.idle()
def set_auto(bot: Bot, update: Update): """ Handles /auto command :param bot: :param update: :return: """ global data chat_id = update.message.chat_id if chat_id in data.conversations: data.conversations[chat_id].auto = not data.conversations[chat_id].auto if data.conversations[chat_id].auto: message = "Automatic mode enabled." else: message = "Automatic mode disabled." send_custom_message(bot=bot, chat_id=chat_id, message=message) else: send_error(bot=bot, chat_id=chat_id, name="account_not_setup")
def restricted(func): @wraps(func) def wrapped(bot, update, *args, **kwargs): # extract user_id from arbitrary update try: user_id = update.message.from_user.id except (NameError, AttributeError): try: user_id = update.inline_query.from_user.id except (NameError, AttributeError): try: user_id = update.chosen_inline_result.from_user.id except (NameError, AttributeError): try: user_id = update.callback_query.from_user.id except (NameError, AttributeError): print("No user_id available in update.") return if user_id not in admin_list: print("Unauthorized access denied for {0}.".format(user_id)) return return func(bot, update, *args, **kwargs) return wrapped # Define a few command handlers. These usually take the two arguments bot and # update. Error handlers also receive the raised TelegramError object in error.
def main(): updater = Updater("INSERT TOKEN HERE") dp = updater.dispatcher dp.add_handler(CommandHandler("start", cmd_start)) dp.add_handler(CommandHandler("md", cmd_markdown, allow_edited=True)) dp.add_handler(CommandHandler("markdown", cmd_markdown, allow_edited=True)) dp.add_handler(CommandHandler("pin", cmd_pin, allow_edited=True)) dp.add_handler(CommandHandler("welcome", cmd_welcome, allow_edited=True)) dp.add_handler(CommandHandler("goodbye", cmd_goodbye, allow_edited=True)) dp.add_handler(CommandHandler("del_welcome", cmd_clear_welcome)) dp.add_handler(CommandHandler("del_goodbye", cmd_clear_goodbye)) dp.add_handler(CommandHandler("set_bot_admin", cmd_set_bot_admin, allow_edited=True)) dp.add_handler(CommandHandler("remove_bot_admin", cmd_remove_bot_admin, allow_edited=True)) dp.add_handler(CommandHandler("settings", cmd_settings)) dp.add_handler(CommandHandler("shortcut", cmd_shortcut_set, allow_edited=True)) dp.add_handler(CommandHandler("del_shortcut", cmd_shortcut_del, allow_edited=True)) dp.add_handler(CommandHandler("shortcuts", cmd_shortcut_getall, allow_edited=True)) dp.add_handler(MessageHandler(Filters.audio | Filters.command | Filters.contact | Filters.document | Filters.photo | Filters.sticker | Filters.text | Filters.video | Filters.voice | Filters.status_update, msg_parse, allow_edited=True)) dp.add_handler(CallbackQueryHandler(inline_button_callback)) dp.add_error_handler(error) updater.start_polling() updater.idle()
def __init__(self, token, generator): self.updater = Updater(token=token) # ??????? ????????? handler = MessageHandler(Filters.text | Filters.command, self.handle_message) self.updater.dispatcher.add_handler(handler) # ?????? ?????????? ???? ????????? ????????? self.handlers = collections.defaultdict(generator) # ??????? ???? "id ???? -> ?????????"
def __init__(self, token, generator, handlers=None): self.updater = Updater(token=token) message_handler = MessageHandler(Filters.text | Filters.command, self.handle_message) inline_query_handler = InlineQueryHandler(self.handle_inline_query) self.updater.dispatcher.add_handler(message_handler) self.updater.dispatcher.add_handler(inline_query_handler) self.generator = generator self.handlers = handlers or {} self.last_message_ids = {}
def unknown_command_callback(self, bot, update): pass # bot.send_message(chat_id=update.message.chat_id, text="Unknown command")
def caps(bot, update, args): text_caps = ' '.join(args).upper() bot.send_message(chat_id=update.message.chat_id, text=text_caps) # displays message to the user if wrong command is initiated # warning : should be added last
def main(): updater = Updater(token=TALKATIVOT_TELEGRAM_TOKEN) # job_queue = updater.job_queue dispatcher = updater.dispatcher dispatcher.add_handler(CommandHandler('start', start)) dispatcher.add_handler(CommandHandler("help", start)) dispatcher.add_handler(MessageHandler(Filters.text, echo)) unknown_handler = MessageHandler(Filters.command, unknown) dispatcher.add_handler(unknown_handler) dispatcher.add_error_handler(error) updater.start_polling() updater.idle()
def __init__(self, tg_bot_token, vk_client_id): self.tg_bot_token = tg_bot_token self.poller = Poller() self.vk = Vk(vk_client_id) self.clients = Client.all_from_db() self.updater = Updater(token=tg_bot_token) dispatcher = self.updater.dispatcher start_command_handler = CommandHandler('start', self.start_command_callback) dispatcher.add_handler(start_command_handler) start_command_handler = CommandHandler('whoami', self.whoami_command_callback) dispatcher.add_handler(start_command_handler) start_command_handler = CommandHandler('pick', self.pick_command_callback) dispatcher.add_handler(start_command_handler) start_command_handler = CommandHandler('unpick', self.unpick_command_callback) dispatcher.add_handler(start_command_handler) start_command_handler = CommandHandler('details', self.details_command_callback) dispatcher.add_handler(start_command_handler) unknown_handler = MessageHandler(Filters.command, self.unknown_command_callback) dispatcher.add_handler(unknown_handler) message_handler = MessageHandler(Filters.text, self.message_callback) dispatcher.add_handler(message_handler) dispatcher.add_error_handler(self.error_callback) self.restore()
def start(bot: Bot, update: Update): """ Handle /start command :param bot: :param update: :return: """ chat_id = update.message.chat_id # TODO if using already logging in -> do not display welcome message send_message(bot, chat_id, "welcome")
def set_location(bot: Bot, update: Update, args): """ Handles /location command :param bot: :param update: :param args: :return: """ global data chat_id = update.message.chat_id if chat_id in data.conversations: if len(args) < 1: send_help(bot, chat_id, "set_location", "Please indicate GPS coordinates or the name of a place") return else: send_chat_action(bot=bot, chat_id=chat_id, action=ChatAction.FIND_LOCATION) r = requests.get("{}{}?format=json&limit=1&bounded=0" .format(location_search_url, ' '.join([str(x) for x in args]))) try: conversation = data.conversations[chat_id] latitude = r.json()[0]["lat"] longitude = r.json()[0]["lon"] conversation.session.update_location(latitude, longitude) send_message(bot, chat_id, "location_updated") conversation.refresh_users() send_location(latitude=latitude, longitude=longitude, bot=bot, chat_id=chat_id) except AttributeError: send_help(bot, chat_id, "set_location", "Facebook token needs to be set first") else: send_error(bot=bot, chat_id=chat_id, name="account_not_setup")
def set_timeout(bot: Bot, update: Update, args): """ Handles /timeout command :param bot: :param update: :param args: :return: """ global data chat_id = update.message.chat_id if chat_id in data.conversations: if len(args) != 1: timeout = str(data.conversations[chat_id].timeout) message = "Current timeout %s seconds." % timeout send_custom_message(bot, chat_id, message=message) else: try: timeout = int(args[0]) settings = data.conversations[chat_id].settings if int(settings.get_setting("min_timeout")) <= timeout <= int( settings.get_setting("max_timeout")): data.conversations[chat_id].timeout = timeout message = "Timeout updated to %d seconds." % data.conversations[chat_id].timeout send_custom_message(bot, chat_id, message=message) else: send_custom_message(bot, chat_id, "Timeout out of range: " + str(settings.get_setting("min_timeout")) + "-" + str(settings.get_setting("max_timeout"))) except AttributeError: message = "An error happened." send_custom_message(bot, chat_id, message=message) except BaseException: message = "An error happened." send_custom_message(bot, chat_id, message=message) else: send_error(bot=bot, chat_id=chat_id, name="account_not_setup")
def start_vote_session(bot: Bot, update: Update, job_queue): """ Handles /new_vote command :param bot: :param update: :param job_queue: :return: """ chat_id = update.message.chat_id job = Job(start_vote, 0, repeat=False, context=(chat_id, job_queue)) job_queue.put(job)
def set_account(bot: Bot, update: Update): """ Handles /set_account command :param bot: :param update: :return: """ global data sender = update.message.from_user.id data.change_account_queries[sender] = update.message.chat_id msg = messages["ask_for_token"] group_name = update.message.chat.title if len(group_name) > 0: msg += " for chat %s" % group_name is_msg_sent = send_private_message(bot, user_id=sender, text=msg) # Check if user already allowed bot to send private messages if not is_msg_sent: notify_start_private_chat(bot=bot, chat_id=data.change_account_queries[sender], incoming_message=update.message) # If the bot is not used in a private chat, display a button allowing user to easily switch to the private chat elif sender != data.change_account_queries[sender]: keyboard = keyboards.switch_private_chat_keyboard(bot.username) notify_send_token(bot=bot, is_group=True, chat_id=data.change_account_queries[sender], reply_to_message_id=update.message.message_id, group_name=group_name, reply_markup=keyboard)
def send_about(bot: Bot, update: Update): """ Send the about message from the /about command :param bot: :param update: :return: """ repo = git.Repo(search_parent_directories=True) sha = repo.head.object.hexsha msg = messages["about"] + "\nLast commit [%s](https://github.com/arthurdk/tinder-telegram-bot/commit/%s)" % ( sha, sha) chat_id = update.message.chat_id send_custom_message(bot=bot, chat_id=chat_id, message=msg)
def custom_command_handler(bot: Bot, update: Update): """ /msg command. Preserves whitespace. Leaves error handling to the chat.send_message method :param bot: :param update: :return: """ if update.message.text.startswith('/msg'): text = update.message.text[4:].strip() if text.startswith("@"): splitter = re.search("\s", text).start() if splitter is not None: text = text[splitter:].strip() splitter = re.search("\s", text).start() if splitter is None: args = [text] else: args = [text[:splitter].strip(), text[splitter:].strip()] chat.send_message(bot, update, args) else: unknown(bot, update)
def main(): db.db.connect() try: db.db.create_tables([db.Conversation, db.User, db.IsMod, db.Vote]) except pw.OperationalError: pass updater = Updater(settings.KEY) dispatcher = updater.dispatcher dispatcher.add_handler(CommandHandler('start', start)) dispatcher.add_handler(CommandHandler('help', send_help_message)) dispatcher.add_handler(CommandHandler('auto', set_auto)) dispatcher.add_handler(CommandHandler('location', set_location, pass_args=True)) dispatcher.add_handler(CommandHandler('set_account', set_account)) dispatcher.add_handler(CommandHandler('unlink', unlink)) dispatcher.add_handler(CommandHandler('matches', send_matches)) dispatcher.add_handler(CallbackQueryHandler(inline.do_press_inline_button, pass_job_queue=True)) dispatcher.add_handler(MessageHandler(Filters.text, message_handler, pass_job_queue=True)) dispatcher.add_handler(MessageHandler(Filters.location, update_location)) dispatcher.add_handler(CommandHandler('new_vote', start_vote_session, pass_job_queue=True)) dispatcher.add_handler(CommandHandler('timeout', set_timeout, pass_args=True)) dispatcher.add_handler(CommandHandler('about', send_about)) # Chat functionality dispatcher.add_handler(CommandHandler('poll_msgs', chat.poll_messages, pass_args=True)) dispatcher.add_handler(CommandHandler('poll_unanswered', chat.poll_unanswered_messages, pass_args=True)) dispatcher.add_handler(CommandHandler('unblock', chat.unblock)) # Settings dispatcher.add_handler(CommandHandler('set_setting', admin.set_setting, pass_args=True)) dispatcher.add_handler(CommandHandler('list_settings', admin.list_settings)) dispatcher.add_handler(CommandHandler('help_settings', admin.help_settings)) # Moderators dispatcher.add_handler(CommandHandler('make_me_a_mod', admin.make_me_a_mod)) inline_caps_handler = InlineQueryHandler(inline.inline_preview) dispatcher.add_handler(inline_caps_handler) dispatcher.add_handler(MessageHandler(Filters.command, custom_command_handler)) dispatcher.add_error_handler(error_callback) updater.start_polling() updater.idle()
