Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
with link_chats(channel, (chat,), bot_group):
await client.send_message(bot_group, "/update_info")
title = (await helper.wait_for_event(in_chats(bot_group) & new_title)).new_title
if alias:
assert chat.alias in title
else:
assert chat.name in title
if avatar:
await helper.wait_for_event(in_chats(bot_group) & new_photo)
if chat_type == "GroupChat":
# Get group description
bot_group_t, peer_type = resolve_id(bot_group)
if peer_type == PeerChannel:
group: ChatFull = await client(GetFullChannelRequest(bot_group_t))
else:
group: ChatFull = await client(GetFullChatRequest(bot_group_t))
desc = group.full_chat.about
chats_found = sum(int(
(i.name in desc) and # Original name is found, and
(i.alias is None or i.alias in desc) # alias is found too if available
) for i in chat.members)
assert len(chat.members) >= 5
assert chats_found >= 5, f"At least 5 members shall be found in the description: {desc}"
if self.fwd_from:
self._forward = Forward(self._client, self.fwd_from, entities)
if self.action:
if isinstance(self.action, (types.MessageActionChatAddUser,
types.MessageActionChatCreate)):
self._action_entities = [entities.get(i)
for i in self.action.users]
elif isinstance(self.action, types.MessageActionChatDeleteUser):
self._action_entities = [entities.get(self.action.user_id)]
elif isinstance(self.action, types.MessageActionChatJoinedByLink):
self._action_entities = [entities.get(self.action.inviter_id)]
elif isinstance(self.action, types.MessageActionChatMigrateTo):
self._action_entities = [entities.get(utils.get_peer_id(
types.PeerChannel(self.action.channel_id)))]
elif isinstance(
self.action, types.MessageActionChannelMigrateFrom):
self._action_entities = [entities.get(utils.get_peer_id(
types.PeerChat(self.action.chat_id)))]
def _set_client(self, client):
if isinstance(self._chat_peer, int):
try:
chat = client._entity_cache[self._chat_peer]
if isinstance(chat, types.InputPeerChat):
self._chat_peer = types.PeerChat(self._chat_peer)
elif isinstance(chat, types.InputPeerChannel):
self._chat_peer = types.PeerChannel(self._chat_peer)
else:
# Should not happen
self._chat_peer = types.PeerUser(self._chat_peer)
except KeyError:
# Hope for the best. We don't know where this event
# occurred but it was most likely in a channel.
self._chat_peer = types.PeerChannel(self._chat_peer)
super()._set_client(client)
self._sender, self._input_sender = utils._get_entity_pair(
self.sender_id, self._entities, client._entity_cache)
def build(cls, update, others=None, self_id=None):
if isinstance(update, types.UpdateChannelPinnedMessage) and update.id == 0:
# Telegram does not always send
# UpdateChannelPinnedMessage for new pins
# but always for unpin, with update.id = 0
return cls.Event(types.PeerChannel(update.channel_id),
unpin=True)
elif isinstance(update, types.UpdateChatParticipantAdd):
return cls.Event(types.PeerChat(update.chat_id),
added_by=update.inviter_id or True,
users=update.user_id)
elif isinstance(update, types.UpdateChatParticipantDelete):
return cls.Event(types.PeerChat(update.chat_id),
kicked_by=True,
users=update.user_id)
elif isinstance(update, types.UpdateChannel):
# We rely on the fact that update._entities is set by _process_update
# This update only has the channel ID, and Telegram *should* have sent
# the entity in the Updates.chats list. If it did, check Channel.left
types.TopPeer, types.Dialog, types.DialogPeer)):
return peer.peer
elif isinstance(peer, types.ChannelFull):
return types.PeerChannel(peer.id)
if peer.SUBCLASS_OF_ID in (0x7d7c6f86, 0xd9c7fc18):
# ChatParticipant, ChannelParticipant
return types.PeerUser(peer.user_id)
peer = get_input_peer(peer, allow_self=False, check_hash=False)
if isinstance(peer, (types.InputPeerUser, types.InputPeerUserFromMessage)):
return types.PeerUser(peer.user_id)
elif isinstance(peer, types.InputPeerChat):
return types.PeerChat(peer.chat_id)
elif isinstance(peer, (types.InputPeerChannel, types.InputPeerChannelFromMessage)):
return types.PeerChannel(peer.channel_id)
except (AttributeError, TypeError):
pass
_raise_cast_fail(peer, 'Peer')
if chats is None:
return None
if not utils.is_list_like(chats):
chats = (chats,)
result = set()
for chat in chats:
if isinstance(chat, int):
if chat < 0:
result.add(chat) # Explicitly marked IDs are negative
else:
result.update({ # Support all valid types of peers
utils.get_peer_id(types.PeerUser(chat)),
utils.get_peer_id(types.PeerChat(chat)),
utils.get_peer_id(types.PeerChannel(chat)),
})
elif isinstance(chat, TLObject) and chat.SUBCLASS_OF_ID == 0x2d45687:
# 0x2d45687 == crc32(b'Peer')
result.add(utils.get_peer_id(chat))
else:
chat = await client.get_input_entity(chat)
if isinstance(chat, types.InputPeerSelf):
chat = await client.get_me(input_peer=True)
result.add(utils.get_peer_id(chat))
return result
def build(cls, update, others=None, self_id=None):
if isinstance(update, types.UpdateReadHistoryInbox):
return cls.Event(update.peer, update.max_id, False)
elif isinstance(update, types.UpdateReadHistoryOutbox):
return cls.Event(update.peer, update.max_id, True)
elif isinstance(update, types.UpdateReadChannelInbox):
return cls.Event(types.PeerChannel(update.channel_id),
update.max_id, False)
elif isinstance(update, types.UpdateReadChannelOutbox):
return cls.Event(types.PeerChannel(update.channel_id),
update.max_id, True)
elif isinstance(update, types.UpdateReadMessagesContents):
return cls.Event(message_ids=update.messages,
contents=True)
elif isinstance(update, types.UpdateChannelReadMessagesContents):
return cls.Event(types.PeerChannel(update.channel_id),
message_ids=update.messages,
contents=True)
def _set_client(self, client):
if isinstance(self._chat_peer, int):
try:
chat = client._entity_cache[self._chat_peer]
if isinstance(chat, types.InputPeerChat):
self._chat_peer = types.PeerChat(self._chat_peer)
elif isinstance(chat, types.InputPeerChannel):
self._chat_peer = types.PeerChannel(self._chat_peer)
else:
# Should not happen
self._chat_peer = types.PeerUser(self._chat_peer)
except KeyError:
# Hope for the best. We don't know where this event
# occurred but it was most likely in a channel.
self._chat_peer = types.PeerChannel(self._chat_peer)
super()._set_client(client)
self._sender, self._input_sender = utils._get_entity_pair(
self.sender_id, self._entities, client._entity_cache)
def build(cls, update, others=None, self_id=None):
if isinstance(update, types.UpdateChannelPinnedMessage) and update.id == 0:
# Telegram does not always send
# UpdateChannelPinnedMessage for new pins
# but always for unpin, with update.id = 0
return cls.Event(types.PeerChannel(update.channel_id),
unpin=True)
elif isinstance(update, types.UpdateChatParticipantAdd):
return cls.Event(types.PeerChat(update.chat_id),
added_by=update.inviter_id or True,
users=update.user_id)
elif isinstance(update, types.UpdateChatParticipantDelete):
return cls.Event(types.PeerChat(update.chat_id),
kicked_by=True,
users=update.user_id)
elif (isinstance(update, (
types.UpdateNewMessage, types.UpdateNewChannelMessage))
and isinstance(update.message, types.MessageService)):
msg = update.message