Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
raise EFBMessageReactionNotPossible("Message reaction is rejected by flag.")
if self.accept_message_reactions == "reject_all":
raise EFBOperationNotSupported("All message reactions are rejected by flag.")
message = self.messages_sent.get(status.msg_id)
if message is None:
raise EFBOperationNotSupported("Message is not found.")
if status.reaction is None:
for idx, i in message.reactions.items():
message.reactions[idx] = [j for j in i if not isinstance(j, SelfChatMember)]
else:
if status.reaction not in message.reactions:
message.reactions[status.reaction] = []
message.reactions[status.reaction].append(message.chat.self)
coordinator.send_status(MessageReactionsUpdate(
chat=message.chat,
msg_id=message.uid,
reactions=message.reactions
))
def send_chat_update_status(self) -> Tuple[Chat, Chat, Chat]:
"""
Returns:
chat added, chat edited, chat removed
"""
keyword = " (Edited)"
if self.backup_chat not in self.chats:
to_add, to_remove = self.backup_chat, self.chat_to_toggle
self.chat_to_edit.name += keyword
else:
to_add, to_remove = self.chat_to_toggle, self.backup_chat
self.chat_to_edit.name = self.chat_to_edit.name.replace(keyword, '')
self.chats.append(to_add)
self.chats.remove(to_remove)
coordinator.send_status(ChatUpdates(
self,
new_chats=[to_add.uid],
modified_chats=[self.chat_to_edit.uid],
removed_chats=[to_remove.uid],
))
return to_add, self.chat_to_edit, to_remove
def send_message_recall_status(self):
slave = next(iter(coordinator.slaves.values()))
alice = slave.get_chat('alice')
msg = EFBMsg()
msg.deliver_to = slave
msg.chat = alice
msg.author = EFBChat(self).self()
msg.uid = "1"
status = EFBMessageRemoval(self, slave, msg)
return coordinator.send_status(status)
msg_md_text = message.text
msg_md_text = msg_md_text or ""
msg_md_caption = message.caption and message.caption_markdown
if msg_md_caption and msg_md_caption == escape_markdown(message.caption):
msg_md_caption = message.caption
msg_md_caption = msg_md_caption or ""
# Flag for edited message
if edited:
m.edit = True
text = msg_md_text or msg_md_caption
m.uid = edited.slave_message_id
if text.startswith(self.DELETE_FLAG):
coordinator.send_status(MessageRemoval(
source_channel=self.channel,
destination_channel=coordinator.slaves[channel],
message=m
))
if not self.channel.flag('prevent_message_removal'):
try:
message.delete()
except TelegramError:
message.reply_text(self._("Message is removed in remote chat."))
else:
message.reply_text(self._("Message is removed in remote chat."))
log_message = False
return
self.logger.debug('[%s] Message is edited (%s)', m.uid, m.edit)
if m.file_unique_id and m.file_unique_id != edited.file_unique_id:
self.logger.debug("[%s] Message media is edited (%s -> %s)", m.uid, edited.file_unique_id, m.file_unique_id)
"This message is not found in ETM database. You cannot remove it from its remote chat."
))
try:
etm_msg: ETMMsg = msg_log.build_etm_msg(self.chat_manager)
except UnpicklingError:
return self.bot.reply_error(update, self._(
"This message is not found in ETM database. You cannot remove it from its remote chat."
))
dest_channel = coordinator.slaves.get(etm_msg.chat.module_id, None)
if dest_channel is None:
return self.bot.reply_error(update, self._(
"Module of this message ({module_id}) could not be found, or is not a slave channel."
).format(module_id=etm_msg.chat.module_id))
# noinspection PyBroadException
try:
coordinator.send_status(MessageRemoval(
source_channel=self.channel, destination_channel=dest_channel, message=etm_msg
))
except EFBException as e:
self.logger.exception("Failed to remove message from remote chat. Message: %s; Error: %s", etm_msg, e)
return reply.reply_text(self._(
"Failed to remove this message from remote chat.\n\n{error!s}"
).format(error=e))
except Exception as e:
self.logger.exception("Failed to remove message from remote chat. Message: %s; Error: %s", etm_msg, e)
return reply.reply_text(self._(
"Failed to remove this message from remote chat.\n\n{error!r}"
).format(error=e))
if not self.channel.flag('prevent_message_removal'):
try:
reply.delete()
except TelegramError:
del self.recall_msg_id_conversion[recall_id]
return None
# val = self.recall_msg_id_conversion.pop(recall_id)
# val[1] -= 1
# if val[1] > 0: # not all associated messages are recalled.
# return None
# else:
# efb_msg.uid = val[0]
else:
# Format message IDs as JSON of List[List[str]].
chat, author = self.get_chat_and_author(msg)
efb_msg = Message(
chat=chat, author=author,
uid=MessageID(json.dumps([[recall_id]]))
)
coordinator.send_status(MessageRemoval(source_channel=self.channel,
destination_channel=coordinator.master,
message=efb_msg))
return None
chat, _ = self.get_chat_and_author(msg)
try:
author = chat.get_member(SystemChatMember.SYSTEM_ID)
except KeyError:
author = chat.add_system_member()
if any(i in msg.text for i in self.NEW_CHAT_PATTERNS):
coordinator.send_status(ChatUpdates(
channel=self.channel,
new_chats=(chat.uid,)
))
elif any(i in msg.text for i in self.CHAT_AND_MEMBER_UPDATE_PATTERNS):
# TODO: detect actual member changes from message text
coordinator.send_status(ChatUpdates(
chat, author = self.get_chat_and_author(msg)
efb_msg = Message(
chat=chat, author=author,
uid=MessageID(json.dumps([[recall_id]]))
)
coordinator.send_status(MessageRemoval(source_channel=self.channel,
destination_channel=coordinator.master,
message=efb_msg))
return None
chat, _ = self.get_chat_and_author(msg)
try:
author = chat.get_member(SystemChatMember.SYSTEM_ID)
except KeyError:
author = chat.add_system_member()
if any(i in msg.text for i in self.NEW_CHAT_PATTERNS):
coordinator.send_status(ChatUpdates(
channel=self.channel,
new_chats=(chat.uid,)
))
elif any(i in msg.text for i in self.CHAT_AND_MEMBER_UPDATE_PATTERNS):
# TODO: detect actual member changes from message text
coordinator.send_status(ChatUpdates(
channel=self.channel,
modified_chats=(chat.uid,)
))
return Message(
text=msg.text,
type=MsgType.Text,
chat=chat,
author=author,
)