def settings_emotion_message(bot: telegram.Bot, update: telegram.Update):
chat_settings = bot_types.ChatSettings.from_db(db, update.message.chat, admin_id=update.message.from_user.id,
admin_name=update.message.from_user.first_name)
if not chat_settings.admin_only or chat_settings.admin_id == update.message.from_user.id:
send_settings_emotion_message(bot, update.message.chat_id)
bot_types.Botan.track(
uid=update.message.chat_id,
message=update.message.to_dict(),
name='settings.emotions.get'
)
logging.info('Command settings: emotion.', extra={
'telegram': {
'update': update.to_dict(),
'chat_id': update.message.chat_id,
'message_id': update.message.message_id,
},
'id': extentions.TextHelper.get_md5(str(update.message.chat_id) + str(update.message.message_id))
})
python类Update()的实例源码
def handle_update(self, update, dispatcher):
"""Send the update to the :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
dispatcher (:class:`telegram.ext.Dispatcher`): Dispatcher that originated the Update.
"""
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)
def check_update(self, update):
"""
Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
Returns:
:obj:`bool`
"""
if isinstance(update, Update) and update.inline_query:
if self.pattern:
if update.inline_query.query:
match = re.match(self.pattern, update.inline_query.query)
return bool(match)
else:
return True
def handle_update(self, update, dispatcher):
"""
Send the update to the :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
dispatcher (:class:`telegram.ext.Dispatcher`): Dispatcher that originated the Update.
"""
optional_args = self.collect_optional_args(dispatcher, update)
if self.pattern:
match = re.match(self.pattern, update.inline_query.query)
if self.pass_groups:
optional_args['groups'] = match.groups()
if self.pass_groupdict:
optional_args['groupdict'] = match.groupdict()
return self.callback(dispatcher.bot, update, **optional_args)
# old non-PEP8 Handler methods
def check_update(self, update):
"""Determines whether an update should be passed to this handlers :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
Returns:
:obj:`bool`
"""
if not isinstance(update, Update) and not update.effective_message:
return False
if any([(self.message_updates and update.message),
(self.edited_updates and update.edited_message),
(self.channel_post_updates and update.channel_post)]) and \
update.effective_message.text:
match = re.match(self.pattern, update.effective_message.text)
return bool(match)
return False
def handle_update(self, update, dispatcher):
"""Send the update to the :attr:`callback`.
Args:
update (:class:`telegram.Update`): Incoming telegram update.
dispatcher (:class:`telegram.ext.Dispatcher`): Dispatcher that originated the Update.
"""
optional_args = self.collect_optional_args(dispatcher, update)
if self.pattern:
match = re.match(self.pattern, update.callback_query.data)
if self.pass_groups:
optional_args['groups'] = match.groups()
if self.pass_groupdict:
optional_args['groupdict'] = match.groupdict()
return self.callback(dispatcher.bot, update, **optional_args)
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 unlink(bot: Bot, update: Update):
"""
Handles /unlink command
:param bot:
:param update:
:return:
"""
global data
sender = update.message.from_user.id
chat_id = update.message.chat_id
if chat_id in data.conversations:
if sender == data.conversations[chat_id].owner:
del data.conversations[chat_id]
send_message(bot, chat_id, name="account_unlinked")
else:
send_error(bot, chat_id=chat_id, name="command_not_allowed")
else:
send_error(bot, chat_id=chat_id, name="account_not_setup")
def _db_update_slave_chats_cache(self, chats):
"""
Update all slave chats info cache to database. Triggered by retrieving
the entire list of chats from all slaves by the method `slave_chats_pagination`.
Args:
chats (list of dict): a list of dicts generated by method `_make_chat_dict`
"""
for i in chats:
db.set_slave_chat_info(slave_channel_id=i['channel_id'],
slave_channel_name=i['channel_name'],
slave_channel_emoji=i['channel_emoji'],
slave_chat_uid=i['chat_uid'],
slave_chat_name=i['chat_name'],
slave_chat_alias=i['chat_alias'],
slave_chat_type=i['type'])
def suggested_recipient(self, bot, chat_id, msg_id, param):
storage_id = "%s.%s" % (chat_id, msg_id)
if param.startswith("chat "):
update = telegram.update.Update.de_json(self.msg_storage[storage_id]["update"], bot)
chat = self.msg_storage[storage_id]['chats'][int(param.split(' ', 1)[1])]
self.process_telegram_message(bot, update, channel_id=chat['channel_id'], chat_id=chat['chat_uid'])
bot.edit_message_text("Delivering the message to %s%s %s: %s." % (chat['channel_emoji'],
utils.Emojis.get_source_emoji(chat['type']),
chat['channel_name'],
chat['chat_name'] if not chat['chat_alias'] or chat['chat_alias'] == chat['chat_name'] else "%s (%s)" % (chat['chat_alias'], chat['chat_name'])),
chat_id=chat_id,
message_id=msg_id)
elif param == Flags.CANCEL_PROCESS:
bot.edit_message_text("Error: No recipient specified.\n"
"Please reply to a previous message.",
chat_id=chat_id,
message_id=msg_id)
else:
bot.edit_message_text("Error: No recipient specified.\n"
"Please reply to a previous message.\n\n"
"Invalid parameter (%s)." % param,
chat_id=chat_id,
message_id=msg_id)
def __get_dummy_update():
"""Returns a dummy update instance"""
user = User(1337, "@foouser")
chat = Chat(1, None)
message = Message(1, user, datetime.now(), chat)
return Update(1, message=message)
def __get_dummy_update():
"""Returns a dummy update instance"""
user = User(1337, "@foouser")
chat = Chat(1, None)
message = Message(1, user, datetime.now(), chat)
return Update(1, message=message)
def command_get_tags(self, bot: Bot, update: Update):
del bot
tags = self.store.get_tags()
tags = [f'` - {tag}`' for tag in tags]
tags = '\n'.join(tags)
text = 'Banned tags:\n' + tags
update.message.reply_text(text, parse_mode=ParseMode.MARKDOWN)
def command_add_tags(self, bot: Bot, update: Update, args: List[str]):
help_msg = 'Usage: `/add_tags <tag> [<tag>]*`'
if len(args) == 0:
update.message.reply_text(help_msg, parse_mode=ParseMode.MARKDOWN)
return
self.store.add_tags(args)
self.command_get_tags(bot, update)
def command_remove_tags(self, bot: Bot, update: Update, args: List[str]):
help_msg = 'Usage: `/remove_tags <tag> [<tag>]*`'
if len(args) == 0:
update.message.reply_text(help_msg, parse_mode=ParseMode.MARKDOWN)
return
self.store.remove_tags(args)
self.command_get_tags(bot, update)
def command_get_settings(self, bot: Bot, update: Update):
del bot
ss = []
for key, value in self.settings.items():
ss.append(f'` - {key} = {value}`')
ss = '\n'.join(ss)
msg = 'Settings list:\n' + ss
update.message.reply_text(msg, parse_mode=ParseMode.MARKDOWN)
def command_change_setting(self, bot: Bot, update: Update, args: List[str]):
help_msg = 'Usage: `/setting <key> <value>`'
if len(args) != 2:
update.message.reply_text(help_msg, parse_mode=ParseMode.MARKDOWN)
return
key, value = args
try:
self.settings.set(key, value)
except SettingsError as e:
update.message.reply_text(str(e))
self.command_get_settings(bot, update)
def command_remove(self, bot: Bot, update: Update, args: List[str]):
if len(args) == 0 or len(args) < 1:
update.message.reply_text('Usage: /remove <subreddit> [<subreddit> ...]')
return
self.store.remove(args)
self.command_list(bot, update)
def command_list(self, bot: Bot, update: Update):
del bot
subreddits = self.store.get()
title = 'subreddit'.ljust(13)
text = f"`{title} limit score`\n"
for name, score in subreddits.items():
text += f'` - {name:10s} {score}`\n'
update.message.reply_text(text, parse_mode=ParseMode.MARKDOWN)
def command_help(self, bot: Bot, update: Update):
text = "/help or /start - print this message.\n" \
"/choose - choose channels.\n" \
"/list - print list of available channels.\n"
if self.chosen_channel:
channel_help = self.chosen_channel.help_text() or ' - no commands'
text += f"\n`{self.chosen_channel.label} ({self.chosen_channel.name})` commands:\n" + channel_help
bot.send_message(chat_id=update.message.chat_id,
text=text)
