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()
python类command()的实例源码
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):
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()
