Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
elif msg.type == MsgType.Status:
# Status messages are not to be recorded in databases
return self.slave_message_status(msg, tg_dest)
elif msg.type == MsgType.Unsupported:
tg_msg = self.slave_message_unsupported(msg, tg_dest, msg_template, reactions, old_msg_id,
target_msg_id, reply_markup, silent)
else:
self.bot.send_chat_action(tg_dest, ChatAction.TYPING)
tg_msg = self.bot.send_message(tg_dest, prefix=msg_template, suffix=reactions,
disable_notification=silent,
text=self._('Unknown type of message "{0}". (UT01)')
.format(msg.type.name))
if tg_msg and commands:
self.channel.commands.register_command(tg_msg, ETMCommandMsgStorage(
commands, coordinator.get_module_by_id(msg.author.module_id), msg_template, msg.text
))
self.logger.debug("[%s] Message is sent to the user with telegram message id %s.%s.",
xid, tg_msg.chat.id, tg_msg.message_id)
etm_msg = ETMMsg.from_efbmsg(msg, self.chat_manager)
etm_msg.type_telegram = get_msg_type(tg_msg)
etm_msg.put_telegram_file(tg_msg)
self.db.add_or_update_message_log(etm_msg, tg_msg, old_msg_id)
# self.logger.debug("[%s] Message inserted/updated to the database.", xid)
def build_link_chats_info_str(self, links: List[EFBChannelChatIDStr]) -> str:
"""Build a string indicating all linked chats in argument.
Returns:
String that starts with a line break.
"""
msg = ""
for i in links:
channel_id, chat_id, _ = etm_utils.chat_id_str_to_id(i)
chat_object = self.chat_manager.get_chat(channel_id, chat_id)
if chat_object:
msg += "\n- %s (%s:%s)" % (chat_object.full_name,
channel_id, chat_id)
else:
try:
module = coordinator.get_module_by_id(channel_id)
if isinstance(module, Channel):
channel_name = f"{module.channel_emoji} {module.channel_name}"
else: # module is Middleware
channel_name = module.middleware_name
msg += self._("\n- {channel_name}: Unknown chat ({channel_id}:{chat_id})").format(
channel_name=channel_name,
channel_id=channel_id,
chat_id=chat_id
)
except NameError:
# TRANSLATORS: ‘channel’ here means an EFB channel.
msg += self._("\n- Unknown channel {channel_id}: ({chat_id})").format(
channel_id=channel_id,
chat_id=chat_id
)
return msg
def __setstate__(self, state: Dict[str, Any]):
self.__dict__.update(state)
with suppress(NameError):
dc = coordinator.get_module_by_id(state['destination_channel'])
if isinstance(dc, EFBChannel):
self.destination_channel = dc
escaped_pattern = re.escape(pattern)
# Use simple string match if no regex significance is found.
if pattern == escaped_pattern:
re_filter = pattern
else:
# Use simple string match if regex provided is invalid
try:
re_filter = re.compile(pattern, re.DOTALL | re.IGNORECASE)
except re.error:
re_filter = pattern
chats: List[ETMChatType] = []
if source_chats:
for s_chat in source_chats:
channel_id, chat_uid, _ = utils.chat_id_str_to_id(s_chat)
with suppress(NameError):
coordinator.get_module_by_id(channel_id)
chat = self.chat_manager.get_chat(channel_id, chat_uid, build_dummy=not filter_availability)
if not chat:
self.logger.debug("slave_chats_pagination with chat list: Chat %s not found.", s_chat)
continue
if chat.match(re_filter):
chats.append(chat)
else:
for etm_chat in self.chat_manager.all_chats:
if etm_chat.match(re_filter):
chats.append(etm_chat)
chats.sort(key=lambda a: a.last_message_time, reverse=True)
chat_list = self.msg_storage[storage_id] = ChatListStorage(chats, offset)
# self._db_update_slave_chats_cache(chat_list.chats)
def __setstate__(self, state: Dict[str, Any]):
self.__dict__.update(state)
# Try to load "deliver_to" channel
with suppress(NameError):
dt = coordinator.get_module_by_id(state['deliver_to'])
if isinstance(dt, EFBChannel):
self.deliver_to = dt
# Try to load file from original path
if self.path:
with suppress(IOError):
self.file = open(self.path, 'rb')
def __setstate__(self, state: Dict[str, Any]):
super(EFBMessageRemoval, self).__setstate__(state)
with suppress(NameError):
sc = coordinator.get_module_by_id(state['source_channel'])
if isinstance(sc, EFBChannel):
self.source_channel = sc
message_id = self.bot.send_message(chat_id, text=self._("Processing...")).message_id
self.bot.send_chat_action(chat_id, ChatAction.TYPING)
if chats and len(chats):
if len(chats) == 1:
slave_channel_id, slave_chat_id, _ = utils.chat_id_str_to_id(chats[0])
# TODO: Channel might be gone, add a check here.
chat = self.chat_manager.get_chat(slave_channel_id, slave_chat_id)
if chat:
msg_text = self._('This group is linked to {0}. '
'Send a message to this group to deliver it to the chat.\n'
'Do NOT reply to this system message.').format(chat.full_name)
else:
try:
channel = coordinator.get_module_by_id(slave_channel_id)
if isinstance(channel, Channel):
name = channel.channel_name
else:
name = channel.middleware_name
msg_text = self._("This group is linked to an unknown chat ({chat_id}) "
"on channel {channel_name} ({channel_id}). Possibly you can "
"no longer reach this chat. Send /unlink_all to unlink all chats "
"from this group.").format(channel_name=name,
channel_id=slave_channel_id,
chat_id=slave_chat_id)
except NameError:
msg_text = self._("This group is linked to a chat from a channel that is not activated "
"({channel_id}, {chat_id}). You cannot reach this chat unless the channel is "
"enabled. Send /unlink_all to unlink all chats "
"from this group.").format(channel_id=slave_channel_id,
chat_id=slave_chat_id)
c_module, c_id, _ = chat_id_str_to_id(self.slave_origin_uid)
a_module, a_id, a_grp = chat_id_str_to_id(self.slave_member_uid)
chat: 'ETMChatType' = chat_manager.get_chat(c_module, c_id, build_dummy=True)
author: 'ETMChatMember' = chat_manager.get_chat_member(a_module, a_grp, a_id, build_dummy=True) # type: ignore
msg = ETMMsg(
uid=self.slave_message_id,
chat=chat,
author=author,
text=self.text,
type=MsgType(self.msg_type),
type_telegram=TGMsgType(self.media_type),
mime=self.mime or None,
file_id=self.file_id or None,
)
with suppress(NameError):
to_module = coordinator.get_module_by_id(self.sent_to)
if isinstance(to_module, Channel):
msg.deliver_to = to_module
# - ``target``: ``master_msg_id`` of the target message
# - ``is_system``
# - ``attributes``
# - ``commands``
# - ``substitutions``: ``Dict[Tuple[int, int], SlaveChatID]``
# - ``reactions``: ``Dict[str, Collection[SlaveChatID]]``
if self.pickle:
misc_data: PickledDict = pickle.loads(self.pickle)
if 'target' in misc_data and recur:
target_row = self.get_or_none(MsgLog.master_msg_id == misc_data['target'])
if target_row:
msg.target = target_row.build_etm_msg(chat_manager, recur=False)
def __init__(self, chat: 'EFBChat', msg_id: str, reaction: Optional[ReactionName]):
"""
Args:
chat (:obj:`EFBChat`): The chat where message is sent
msg_id (str): ID of the message to react to
reaction (Optional[str]): The reaction name to be sent, usually an emoji
"""
self.chat: 'EFBChat' = chat
self.msg_id: str = msg_id
self.reaction: Optional[str] = reaction
if getattr(self.chat, 'module_id', None):
dc = coordinator.get_module_by_id(self.chat.module_id)
if isinstance(dc, EFBChannel):
self.destination_channel = dc
self.verify()