1062 lines
40 KiB
Python
1062 lines
40 KiB
Python
# Pyrogram - Telegram MTProto API Client Library for Python
|
|
# Copyright (C) 2017-2018 Dan Tès <https://github.com/delivrance>
|
|
#
|
|
# This file is part of Pyrogram.
|
|
#
|
|
# Pyrogram is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Lesser General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# Pyrogram is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Lesser General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Lesser General Public License
|
|
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import logging
|
|
from base64 import b64decode, b64encode
|
|
from struct import pack
|
|
|
|
from pyrogram.client import types as pyrogram_types
|
|
from ...api import types, functions
|
|
from ...api.errors import StickersetInvalid, MessageIdsEmpty
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# TODO: Organize the code better?
|
|
|
|
class Str(str):
|
|
__slots__ = "_client", "_entities"
|
|
|
|
def __init__(self, *args):
|
|
super().__init__()
|
|
self._client = None
|
|
self._entities = None
|
|
|
|
def init(self, client, entities):
|
|
self._client = client
|
|
self._entities = entities
|
|
|
|
@property
|
|
def text(self):
|
|
return self
|
|
|
|
@property
|
|
def markdown(self):
|
|
return self._client.markdown.unparse(self, self._entities)
|
|
|
|
@property
|
|
def html(self):
|
|
return self._client.html.unparse(self, self._entities)
|
|
|
|
|
|
ENTITIES = {
|
|
types.MessageEntityMention.ID: "mention",
|
|
types.MessageEntityHashtag.ID: "hashtag",
|
|
types.MessageEntityCashtag.ID: "cashtag",
|
|
types.MessageEntityBotCommand.ID: "bot_command",
|
|
types.MessageEntityUrl.ID: "url",
|
|
types.MessageEntityEmail.ID: "email",
|
|
types.MessageEntityBold.ID: "bold",
|
|
types.MessageEntityItalic.ID: "italic",
|
|
types.MessageEntityCode.ID: "code",
|
|
types.MessageEntityPre.ID: "pre",
|
|
types.MessageEntityTextUrl.ID: "text_link",
|
|
types.MessageEntityMentionName.ID: "text_mention",
|
|
types.MessageEntityPhone.ID: "phone_number"
|
|
}
|
|
|
|
|
|
def parse_entities(entities: list, users: dict) -> list:
|
|
output_entities = []
|
|
|
|
for entity in entities:
|
|
entity_type = ENTITIES.get(entity.ID, None)
|
|
|
|
if entity_type:
|
|
output_entities.append(
|
|
pyrogram_types.MessageEntity(
|
|
type=entity_type,
|
|
offset=entity.offset,
|
|
length=entity.length,
|
|
url=getattr(entity, "url", None),
|
|
user=parse_user(
|
|
users.get(
|
|
getattr(entity, "user_id", None),
|
|
None
|
|
)
|
|
)
|
|
)
|
|
)
|
|
|
|
return output_entities
|
|
|
|
|
|
def parse_chat_photo(photo):
|
|
if not isinstance(photo, (types.UserProfilePhoto, types.ChatPhoto)):
|
|
return None
|
|
|
|
if not isinstance(photo.photo_small, types.FileLocation):
|
|
return None
|
|
|
|
if not isinstance(photo.photo_big, types.FileLocation):
|
|
return None
|
|
|
|
photo_id = getattr(photo, "photo_id", 0)
|
|
loc_small = photo.photo_small
|
|
loc_big = photo.photo_big
|
|
|
|
return pyrogram_types.ChatPhoto(
|
|
small_file_id=encode(
|
|
pack(
|
|
"<iiqqqqi", 1, loc_small.dc_id, photo_id, 0, loc_small.volume_id,
|
|
loc_small.secret, loc_small.local_id
|
|
)
|
|
),
|
|
big_file_id=encode(
|
|
pack(
|
|
"<iiqqqqi", 1, loc_big.dc_id, photo_id, 0, loc_big.volume_id,
|
|
loc_big.secret, loc_big.local_id
|
|
)
|
|
)
|
|
)
|
|
|
|
|
|
def parse_user_status(user_status, user_id: int = None, is_bot: bool = False) -> pyrogram_types.UserStatus or None:
|
|
if is_bot:
|
|
return None
|
|
|
|
status = pyrogram_types.UserStatus(user_id)
|
|
|
|
if isinstance(user_status, types.UserStatusOnline):
|
|
status.online = True
|
|
status.date = user_status.expires
|
|
elif isinstance(user_status, types.UserStatusOffline):
|
|
status.offline = True
|
|
status.date = user_status.was_online
|
|
elif isinstance(user_status, types.UserStatusRecently):
|
|
status.recently = True
|
|
elif isinstance(user_status, types.UserStatusLastWeek):
|
|
status.within_week = True
|
|
elif isinstance(user_status, types.UserStatusLastMonth):
|
|
status.within_month = True
|
|
else:
|
|
status.long_time_ago = True
|
|
|
|
return status
|
|
|
|
|
|
def parse_user(user: types.User) -> pyrogram_types.User or None:
|
|
return pyrogram_types.User(
|
|
id=user.id,
|
|
is_self=user.is_self,
|
|
is_contact=user.contact,
|
|
is_mutual_contact=user.mutual_contact,
|
|
is_deleted=user.deleted,
|
|
is_bot=user.bot,
|
|
first_name=user.first_name,
|
|
last_name=user.last_name,
|
|
username=user.username,
|
|
language_code=user.lang_code,
|
|
phone_number=user.phone,
|
|
photo=parse_chat_photo(user.photo),
|
|
status=parse_user_status(user.status, is_bot=user.bot),
|
|
restriction_reason=user.restriction_reason
|
|
) if user else None
|
|
|
|
|
|
def parse_chat(message: types.Message, users: dict, chats: dict) -> pyrogram_types.Chat:
|
|
if isinstance(message.to_id, types.PeerUser):
|
|
return parse_user_chat(users[message.to_id.user_id if message.out else message.from_id])
|
|
elif isinstance(message.to_id, types.PeerChat):
|
|
return parse_chat_chat(chats[message.to_id.chat_id])
|
|
else:
|
|
return parse_channel_chat(chats[message.to_id.channel_id])
|
|
|
|
|
|
def parse_user_chat(user: types.User) -> pyrogram_types.Chat:
|
|
return pyrogram_types.Chat(
|
|
id=user.id,
|
|
type="private",
|
|
username=user.username,
|
|
first_name=user.first_name,
|
|
last_name=user.last_name,
|
|
photo=parse_chat_photo(user.photo),
|
|
restriction_reason=user.restriction_reason
|
|
)
|
|
|
|
|
|
def parse_chat_chat(chat: types.Chat) -> pyrogram_types.Chat:
|
|
admins_enabled = getattr(chat, "admins_enabled", None)
|
|
|
|
if admins_enabled is not None:
|
|
admins_enabled = not admins_enabled
|
|
|
|
return pyrogram_types.Chat(
|
|
id=-chat.id,
|
|
type="group",
|
|
title=chat.title,
|
|
all_members_are_administrators=admins_enabled,
|
|
photo=parse_chat_photo(getattr(chat, "photo", None))
|
|
)
|
|
|
|
|
|
def parse_channel_chat(channel: types.Channel) -> pyrogram_types.Chat:
|
|
return pyrogram_types.Chat(
|
|
id=int("-100" + str(channel.id)),
|
|
type="supergroup" if channel.megagroup else "channel",
|
|
title=channel.title,
|
|
username=getattr(channel, "username", None),
|
|
photo=parse_chat_photo(getattr(channel, "photo", None)),
|
|
restriction_reason=getattr(channel, "restriction_reason", None)
|
|
)
|
|
|
|
|
|
def parse_thumb(thumb: types.PhotoSize or types.PhotoCachedSize) -> pyrogram_types.PhotoSize or None:
|
|
if isinstance(thumb, (types.PhotoSize, types.PhotoCachedSize)):
|
|
loc = thumb.location
|
|
|
|
if isinstance(thumb, types.PhotoSize):
|
|
file_size = thumb.size
|
|
else:
|
|
file_size = len(thumb.bytes)
|
|
|
|
if isinstance(loc, types.FileLocation):
|
|
return pyrogram_types.PhotoSize(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqqqqi",
|
|
0,
|
|
loc.dc_id,
|
|
0,
|
|
0,
|
|
loc.volume_id,
|
|
loc.secret,
|
|
loc.local_id
|
|
)
|
|
),
|
|
width=thumb.w,
|
|
height=thumb.h,
|
|
file_size=file_size
|
|
)
|
|
|
|
|
|
def decode(s: str) -> bytes:
|
|
s = b64decode(s + "=" * (-len(s) % 4), "-_")
|
|
r = b""
|
|
|
|
assert s[-1] == 2
|
|
|
|
i = 0
|
|
while i < len(s) - 1:
|
|
if s[i] != 0:
|
|
r += bytes([s[i]])
|
|
else:
|
|
r += b"\x00" * s[i + 1]
|
|
i += 1
|
|
|
|
i += 1
|
|
|
|
return r
|
|
|
|
|
|
def encode(s: bytes) -> str:
|
|
r = b""
|
|
n = 0
|
|
|
|
for i in s + bytes([2]):
|
|
if i == 0:
|
|
n += 1
|
|
else:
|
|
if n:
|
|
r += b"\x00" + bytes([n])
|
|
n = 0
|
|
|
|
r += bytes([i])
|
|
|
|
return b64encode(r, b"-_").decode().rstrip("=")
|
|
|
|
|
|
# TODO: Reorganize code, maybe split parts as well
|
|
def parse_messages(
|
|
client,
|
|
messages: list or types.Message or types.MessageService or types.MessageEmpty,
|
|
users: dict,
|
|
chats: dict,
|
|
replies: int = 1
|
|
) -> pyrogram_types.Message or list:
|
|
is_list = isinstance(messages, list)
|
|
messages = messages if is_list else [messages]
|
|
parsed_messages = []
|
|
|
|
for message in messages:
|
|
if isinstance(message, types.Message):
|
|
entities = parse_entities(message.entities, users)
|
|
|
|
forward_from = None
|
|
forward_from_chat = None
|
|
forward_from_message_id = None
|
|
forward_signature = None
|
|
forward_date = None
|
|
|
|
forward_header = message.fwd_from # type: types.MessageFwdHeader
|
|
|
|
if forward_header:
|
|
forward_date = forward_header.date
|
|
|
|
if forward_header.from_id:
|
|
forward_from = parse_user(users[forward_header.from_id])
|
|
else:
|
|
forward_from_chat = parse_channel_chat(chats[forward_header.channel_id])
|
|
forward_from_message_id = forward_header.channel_post
|
|
forward_signature = forward_header.post_author
|
|
|
|
photo = None
|
|
location = None
|
|
contact = None
|
|
venue = None
|
|
audio = None
|
|
voice = None
|
|
animation = None
|
|
video = None
|
|
video_note = None
|
|
sticker = None
|
|
document = None
|
|
web_page = None
|
|
|
|
media = message.media
|
|
|
|
if media:
|
|
if isinstance(media, types.MessageMediaPhoto):
|
|
photo = media.photo
|
|
|
|
if isinstance(photo, types.Photo):
|
|
sizes = photo.sizes
|
|
photo_sizes = []
|
|
|
|
for size in sizes:
|
|
if isinstance(size, (types.PhotoSize, types.PhotoCachedSize)):
|
|
loc = size.location
|
|
|
|
if isinstance(size, types.PhotoSize):
|
|
file_size = size.size
|
|
else:
|
|
file_size = len(size.bytes)
|
|
|
|
if isinstance(loc, types.FileLocation):
|
|
photo_size = pyrogram_types.PhotoSize(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqqqqi",
|
|
2,
|
|
loc.dc_id,
|
|
photo.id,
|
|
photo.access_hash,
|
|
loc.volume_id,
|
|
loc.secret,
|
|
loc.local_id
|
|
)
|
|
),
|
|
width=size.w,
|
|
height=size.h,
|
|
file_size=file_size
|
|
)
|
|
|
|
photo_sizes.append(photo_size)
|
|
|
|
photo = pyrogram_types.Photo(
|
|
id=b64encode(
|
|
pack(
|
|
"<qq",
|
|
photo.id,
|
|
photo.access_hash
|
|
),
|
|
b"-_"
|
|
).decode().rstrip("="),
|
|
date=photo.date,
|
|
sizes=photo_sizes
|
|
)
|
|
elif isinstance(media, types.MessageMediaGeo):
|
|
geo_point = media.geo
|
|
|
|
if isinstance(geo_point, types.GeoPoint):
|
|
location = pyrogram_types.Location(
|
|
longitude=geo_point.long,
|
|
latitude=geo_point.lat
|
|
)
|
|
elif isinstance(media, types.MessageMediaContact):
|
|
contact = pyrogram_types.Contact(
|
|
phone_number=media.phone_number,
|
|
first_name=media.first_name,
|
|
last_name=media.last_name or None,
|
|
vcard=media.vcard or None,
|
|
user_id=media.user_id or None
|
|
)
|
|
elif isinstance(media, types.MessageMediaVenue):
|
|
venue = pyrogram_types.Venue(
|
|
location=pyrogram_types.Location(
|
|
longitude=media.geo.long,
|
|
latitude=media.geo.lat
|
|
),
|
|
title=media.title,
|
|
address=media.address,
|
|
foursquare_id=media.venue_id or None,
|
|
foursquare_type=media.venue_type
|
|
)
|
|
elif isinstance(media, types.MessageMediaDocument):
|
|
doc = media.document
|
|
|
|
if isinstance(doc, types.Document):
|
|
attributes = {type(i): i for i in doc.attributes}
|
|
|
|
file_name = getattr(
|
|
attributes.get(
|
|
types.DocumentAttributeFilename, None
|
|
), "file_name", None
|
|
)
|
|
|
|
if types.DocumentAttributeAudio in attributes:
|
|
audio_attributes = attributes[types.DocumentAttributeAudio]
|
|
|
|
if audio_attributes.voice:
|
|
voice = pyrogram_types.Voice(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqq",
|
|
3,
|
|
doc.dc_id,
|
|
doc.id,
|
|
doc.access_hash
|
|
)
|
|
),
|
|
duration=audio_attributes.duration,
|
|
mime_type=doc.mime_type,
|
|
file_size=doc.size,
|
|
waveform=audio_attributes.waveform,
|
|
date=doc.date
|
|
)
|
|
else:
|
|
audio = pyrogram_types.Audio(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqq",
|
|
9,
|
|
doc.dc_id,
|
|
doc.id,
|
|
doc.access_hash
|
|
)
|
|
),
|
|
duration=audio_attributes.duration,
|
|
performer=audio_attributes.performer,
|
|
title=audio_attributes.title,
|
|
mime_type=doc.mime_type,
|
|
file_size=doc.size,
|
|
thumb=parse_thumb(doc.thumb),
|
|
file_name=file_name,
|
|
date=doc.date
|
|
)
|
|
elif types.DocumentAttributeAnimated in attributes:
|
|
video_attributes = attributes.get(types.DocumentAttributeVideo, None)
|
|
|
|
animation = pyrogram_types.Animation(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqq",
|
|
10,
|
|
doc.dc_id,
|
|
doc.id,
|
|
doc.access_hash
|
|
)
|
|
),
|
|
width=getattr(video_attributes, "w", 0),
|
|
height=getattr(video_attributes, "h", 0),
|
|
duration=getattr(video_attributes, "duration", 0),
|
|
thumb=parse_thumb(doc.thumb),
|
|
mime_type=doc.mime_type,
|
|
file_size=doc.size,
|
|
file_name=file_name,
|
|
date=doc.date
|
|
)
|
|
elif types.DocumentAttributeVideo in attributes:
|
|
video_attributes = attributes[types.DocumentAttributeVideo]
|
|
|
|
if video_attributes.round_message:
|
|
video_note = pyrogram_types.VideoNote(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqq",
|
|
13,
|
|
doc.dc_id,
|
|
doc.id,
|
|
doc.access_hash
|
|
)
|
|
),
|
|
length=video_attributes.w,
|
|
duration=video_attributes.duration,
|
|
thumb=parse_thumb(doc.thumb),
|
|
file_size=doc.size,
|
|
mime_type=doc.mime_type,
|
|
date=doc.date
|
|
)
|
|
else:
|
|
video = pyrogram_types.Video(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqq",
|
|
4,
|
|
doc.dc_id,
|
|
doc.id,
|
|
doc.access_hash
|
|
)
|
|
),
|
|
width=video_attributes.w,
|
|
height=video_attributes.h,
|
|
duration=video_attributes.duration,
|
|
thumb=parse_thumb(doc.thumb),
|
|
mime_type=doc.mime_type,
|
|
file_size=doc.size,
|
|
file_name=file_name,
|
|
date=doc.date
|
|
)
|
|
elif types.DocumentAttributeSticker in attributes:
|
|
image_size_attributes = attributes.get(types.DocumentAttributeImageSize, None)
|
|
sticker_attribute = attributes[types.DocumentAttributeSticker]
|
|
|
|
if isinstance(sticker_attribute.stickerset, types.InputStickerSetID):
|
|
try:
|
|
set_name = client.send(
|
|
functions.messages.GetStickerSet(sticker_attribute.stickerset)
|
|
).set.short_name
|
|
except StickersetInvalid:
|
|
set_name = None
|
|
else:
|
|
set_name = None
|
|
|
|
sticker = pyrogram_types.Sticker(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqq",
|
|
8,
|
|
doc.dc_id,
|
|
doc.id,
|
|
doc.access_hash
|
|
)
|
|
),
|
|
width=image_size_attributes.w if image_size_attributes else 0,
|
|
height=image_size_attributes.h if image_size_attributes else 0,
|
|
thumb=parse_thumb(doc.thumb),
|
|
# TODO: mask_position
|
|
set_name=set_name,
|
|
emoji=sticker_attribute.alt or None,
|
|
file_size=doc.size,
|
|
mime_type=doc.mime_type,
|
|
file_name=file_name,
|
|
date=doc.date
|
|
)
|
|
else:
|
|
document = pyrogram_types.Document(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqq",
|
|
5,
|
|
doc.dc_id,
|
|
doc.id,
|
|
doc.access_hash
|
|
)
|
|
),
|
|
thumb=parse_thumb(doc.thumb),
|
|
file_name=file_name,
|
|
mime_type=doc.mime_type,
|
|
file_size=doc.size,
|
|
date=doc.date
|
|
)
|
|
elif isinstance(media, types.MessageMediaWebPage):
|
|
web_page = True
|
|
else:
|
|
media = None
|
|
|
|
reply_markup = message.reply_markup
|
|
|
|
if reply_markup:
|
|
if isinstance(reply_markup, types.ReplyKeyboardForceReply):
|
|
reply_markup = pyrogram_types.ForceReply.read(reply_markup)
|
|
elif isinstance(reply_markup, types.ReplyKeyboardMarkup):
|
|
reply_markup = pyrogram_types.ReplyKeyboardMarkup.read(reply_markup)
|
|
elif isinstance(reply_markup, types.ReplyInlineMarkup):
|
|
reply_markup = pyrogram_types.InlineKeyboardMarkup.read(reply_markup)
|
|
elif isinstance(reply_markup, types.ReplyKeyboardHide):
|
|
reply_markup = pyrogram_types.ReplyKeyboardRemove.read(reply_markup)
|
|
else:
|
|
reply_markup = None
|
|
|
|
m = pyrogram_types.Message(
|
|
message_id=message.id,
|
|
date=message.date,
|
|
chat=parse_chat(message, users, chats),
|
|
from_user=parse_user(users.get(message.from_id, None)),
|
|
text=Str(message.message) or None if media is None else None,
|
|
caption=Str(message.message) or None if media is not None else None,
|
|
entities=entities or None if media is None else None,
|
|
caption_entities=entities or None if media is not None else None,
|
|
author_signature=message.post_author,
|
|
forward_from=forward_from,
|
|
forward_from_chat=forward_from_chat,
|
|
forward_from_message_id=forward_from_message_id,
|
|
forward_signature=forward_signature,
|
|
forward_date=forward_date,
|
|
mentioned=message.mentioned,
|
|
media=bool(media) or None,
|
|
edit_date=message.edit_date,
|
|
media_group_id=message.grouped_id,
|
|
photo=photo,
|
|
location=location,
|
|
contact=contact,
|
|
venue=venue,
|
|
audio=audio,
|
|
voice=voice,
|
|
animation=animation,
|
|
video=video,
|
|
video_note=video_note,
|
|
sticker=sticker,
|
|
document=document,
|
|
web_page=web_page,
|
|
views=message.views,
|
|
via_bot=parse_user(users.get(message.via_bot_id, None)),
|
|
outgoing=message.out,
|
|
client=client,
|
|
reply_markup=reply_markup
|
|
)
|
|
|
|
if m.text:
|
|
m.text.init(m._client, m.entities or [])
|
|
|
|
if m.caption:
|
|
m.caption.init(m._client, m.caption_entities or [])
|
|
|
|
if message.reply_to_msg_id and replies:
|
|
try:
|
|
m.reply_to_message = client.get_messages(
|
|
m.chat.id,
|
|
reply_to_message_ids=message.id,
|
|
replies=replies - 1
|
|
)
|
|
except MessageIdsEmpty:
|
|
pass
|
|
elif isinstance(message, types.MessageService):
|
|
action = message.action
|
|
|
|
new_chat_members = None
|
|
left_chat_member = None
|
|
new_chat_title = None
|
|
delete_chat_photo = None
|
|
migrate_to_chat_id = None
|
|
migrate_from_chat_id = None
|
|
group_chat_created = None
|
|
channel_chat_created = None
|
|
new_chat_photo = None
|
|
|
|
if isinstance(action, types.MessageActionChatAddUser):
|
|
new_chat_members = [parse_user(users[i]) for i in action.users]
|
|
elif isinstance(action, types.MessageActionChatJoinedByLink):
|
|
new_chat_members = [parse_user(users[message.from_id])]
|
|
elif isinstance(action, types.MessageActionChatDeleteUser):
|
|
left_chat_member = parse_user(users[action.user_id])
|
|
elif isinstance(action, types.MessageActionChatEditTitle):
|
|
new_chat_title = action.title
|
|
elif isinstance(action, types.MessageActionChatDeletePhoto):
|
|
delete_chat_photo = True
|
|
elif isinstance(action, types.MessageActionChatMigrateTo):
|
|
migrate_to_chat_id = action.channel_id
|
|
elif isinstance(action, types.MessageActionChannelMigrateFrom):
|
|
migrate_from_chat_id = action.chat_id
|
|
elif isinstance(action, types.MessageActionChatCreate):
|
|
group_chat_created = True
|
|
elif isinstance(action, types.MessageActionChannelCreate):
|
|
channel_chat_created = True
|
|
elif isinstance(action, types.MessageActionChatEditPhoto):
|
|
photo = action.photo
|
|
|
|
if isinstance(photo, types.Photo):
|
|
sizes = photo.sizes
|
|
photo_sizes = []
|
|
|
|
for size in sizes:
|
|
if isinstance(size, (types.PhotoSize, types.PhotoCachedSize)):
|
|
loc = size.location
|
|
|
|
if isinstance(size, types.PhotoSize):
|
|
file_size = size.size
|
|
else:
|
|
file_size = len(size.bytes)
|
|
|
|
if isinstance(loc, types.FileLocation):
|
|
photo_size = pyrogram_types.PhotoSize(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqqqqi",
|
|
2,
|
|
loc.dc_id,
|
|
photo.id,
|
|
photo.access_hash,
|
|
loc.volume_id,
|
|
loc.secret,
|
|
loc.local_id
|
|
)
|
|
),
|
|
width=size.w,
|
|
height=size.h,
|
|
file_size=file_size
|
|
)
|
|
|
|
photo_sizes.append(photo_size)
|
|
|
|
new_chat_photo = pyrogram_types.Photo(
|
|
id=b64encode(
|
|
pack(
|
|
"<qq",
|
|
photo.id,
|
|
photo.access_hash
|
|
),
|
|
b"-_"
|
|
).decode().rstrip("="),
|
|
date=photo.date,
|
|
sizes=photo_sizes
|
|
)
|
|
|
|
m = pyrogram_types.Message(
|
|
message_id=message.id,
|
|
date=message.date,
|
|
chat=parse_chat(message, users, chats),
|
|
from_user=parse_user(users.get(message.from_id, None)),
|
|
service=True,
|
|
new_chat_members=new_chat_members,
|
|
left_chat_member=left_chat_member,
|
|
new_chat_title=new_chat_title,
|
|
new_chat_photo=new_chat_photo,
|
|
delete_chat_photo=delete_chat_photo,
|
|
migrate_to_chat_id=int("-100" + str(migrate_to_chat_id)) if migrate_to_chat_id else None,
|
|
migrate_from_chat_id=-migrate_from_chat_id if migrate_from_chat_id else None,
|
|
group_chat_created=group_chat_created,
|
|
channel_chat_created=channel_chat_created,
|
|
client=client
|
|
# TODO: supergroup_chat_created
|
|
)
|
|
|
|
if isinstance(action, types.MessageActionPinMessage):
|
|
try:
|
|
m.pinned_message = client.get_messages(
|
|
m.chat.id,
|
|
reply_to_message_ids=message.id,
|
|
replies=0
|
|
)
|
|
except MessageIdsEmpty:
|
|
pass
|
|
else:
|
|
m = pyrogram_types.Message(message_id=message.id, client=client, empty=True)
|
|
|
|
parsed_messages.append(m)
|
|
|
|
return parsed_messages if is_list else parsed_messages[0]
|
|
|
|
|
|
def parse_deleted_messages(update) -> pyrogram_types.Messages:
|
|
messages = update.messages
|
|
channel_id = getattr(update, "channel_id", None)
|
|
|
|
parsed_messages = []
|
|
|
|
for message in messages:
|
|
parsed_messages.append(
|
|
pyrogram_types.Message(
|
|
message_id=message,
|
|
chat=(pyrogram_types.Chat(id=int("-100" + str(channel_id)), type="channel")
|
|
if channel_id is not None
|
|
else None)
|
|
)
|
|
)
|
|
|
|
return pyrogram_types.Messages(len(parsed_messages), parsed_messages)
|
|
|
|
|
|
def get_peer_id(input_peer) -> int:
|
|
return (
|
|
input_peer.user_id if isinstance(input_peer, types.InputPeerUser)
|
|
else -input_peer.chat_id if isinstance(input_peer, types.InputPeerChat)
|
|
else int("-100" + str(input_peer.channel_id))
|
|
)
|
|
|
|
|
|
def get_input_peer(peer_id: int, access_hash: int):
|
|
return (
|
|
types.InputPeerUser(peer_id, access_hash) if peer_id > 0
|
|
else types.InputPeerChannel(int(str(peer_id)[4:]), access_hash)
|
|
if (str(peer_id).startswith("-100") and access_hash)
|
|
else types.InputPeerChat(-peer_id)
|
|
)
|
|
|
|
|
|
def get_offset_date(dialogs):
|
|
for m in reversed(dialogs.messages):
|
|
if isinstance(m, types.MessageEmpty):
|
|
continue
|
|
else:
|
|
return m.date
|
|
else:
|
|
return 0
|
|
|
|
|
|
def parse_profile_photos(photos):
|
|
if isinstance(photos, types.photos.Photos):
|
|
total_count = len(photos.photos)
|
|
else:
|
|
total_count = photos.count
|
|
|
|
user_profile_photos = []
|
|
|
|
for photo in photos.photos:
|
|
if isinstance(photo, types.Photo):
|
|
sizes = photo.sizes
|
|
photo_sizes = []
|
|
|
|
for size in sizes:
|
|
if isinstance(size, (types.PhotoSize, types.PhotoCachedSize)):
|
|
loc = size.location
|
|
|
|
if isinstance(size, types.PhotoSize):
|
|
file_size = size.size
|
|
else:
|
|
file_size = len(size.bytes)
|
|
|
|
if isinstance(loc, types.FileLocation):
|
|
photo_size = pyrogram_types.PhotoSize(
|
|
file_id=encode(
|
|
pack(
|
|
"<iiqqqqi",
|
|
2,
|
|
loc.dc_id,
|
|
photo.id,
|
|
photo.access_hash,
|
|
loc.volume_id,
|
|
loc.secret,
|
|
loc.local_id
|
|
)
|
|
),
|
|
width=size.w,
|
|
height=size.h,
|
|
file_size=file_size
|
|
)
|
|
|
|
photo_sizes.append(photo_size)
|
|
|
|
user_profile_photos.append(
|
|
pyrogram_types.Photo(
|
|
id=b64encode(
|
|
pack(
|
|
"<qq",
|
|
photo.id,
|
|
photo.access_hash
|
|
),
|
|
b"-_"
|
|
).decode().rstrip("="),
|
|
date=photo.date,
|
|
sizes=photo_sizes
|
|
)
|
|
)
|
|
|
|
return pyrogram_types.UserProfilePhotos(
|
|
total_count=total_count,
|
|
photos=user_profile_photos
|
|
)
|
|
|
|
|
|
def parse_callback_query(client, update, users):
|
|
message = None
|
|
inline_message_id = None
|
|
|
|
if isinstance(update, types.UpdateBotCallbackQuery):
|
|
peer = update.peer
|
|
|
|
if isinstance(peer, types.PeerUser):
|
|
peer_id = peer.user_id
|
|
elif isinstance(peer, types.PeerChat):
|
|
peer_id = -peer.chat_id
|
|
else:
|
|
peer_id = int("-100" + str(peer.channel_id))
|
|
|
|
message = client.get_messages(peer_id, update.msg_id)
|
|
elif isinstance(update, types.UpdateInlineBotCallbackQuery):
|
|
inline_message_id = b64encode(
|
|
pack(
|
|
"<iqq",
|
|
update.msg_id.dc_id,
|
|
update.msg_id.id,
|
|
update.msg_id.access_hash
|
|
),
|
|
b"-_"
|
|
).decode().rstrip("=")
|
|
|
|
return pyrogram_types.CallbackQuery(
|
|
id=str(update.query_id),
|
|
from_user=parse_user(users[update.user_id]),
|
|
message=message,
|
|
inline_message_id=inline_message_id,
|
|
chat_instance=str(update.chat_instance),
|
|
data=update.data.decode(),
|
|
game_short_name=update.game_short_name,
|
|
client=client
|
|
)
|
|
|
|
|
|
def parse_chat_full(
|
|
client,
|
|
chat_full: types.messages.ChatFull or types.UserFull
|
|
) -> pyrogram_types.Chat:
|
|
if isinstance(chat_full, types.UserFull):
|
|
parsed_chat = parse_user_chat(chat_full.user)
|
|
parsed_chat.description = chat_full.about
|
|
else:
|
|
full_chat = chat_full.full_chat
|
|
chat = None
|
|
|
|
for i in chat_full.chats:
|
|
if full_chat.id == i.id:
|
|
chat = i
|
|
|
|
if isinstance(full_chat, types.ChatFull):
|
|
parsed_chat = parse_chat_chat(chat)
|
|
|
|
if isinstance(full_chat.participants, types.ChatParticipants):
|
|
parsed_chat.members_count = len(full_chat.participants.participants)
|
|
else:
|
|
parsed_chat = parse_channel_chat(chat)
|
|
parsed_chat.members_count = full_chat.participants_count
|
|
parsed_chat.description = full_chat.about or None
|
|
# TODO: Add StickerSet type
|
|
parsed_chat.can_set_sticker_set = full_chat.can_set_stickers
|
|
parsed_chat.sticker_set_name = full_chat.stickerset
|
|
|
|
if full_chat.pinned_msg_id:
|
|
parsed_chat.pinned_message = client.get_messages(
|
|
parsed_chat.id,
|
|
message_ids=full_chat.pinned_msg_id
|
|
)
|
|
|
|
if isinstance(full_chat.exported_invite, types.ChatInviteExported):
|
|
parsed_chat.invite_link = full_chat.exported_invite.link
|
|
|
|
return parsed_chat
|
|
|
|
|
|
def parse_dialog_chat(peer, users: dict, chats: dict):
|
|
if isinstance(peer, types.PeerUser):
|
|
return parse_user_chat(users[peer.user_id])
|
|
elif isinstance(peer, types.PeerChat):
|
|
return parse_chat_chat(chats[peer.chat_id])
|
|
else:
|
|
return parse_channel_chat(chats[peer.channel_id])
|
|
|
|
|
|
def parse_chat_members(members: types.channels.ChannelParticipants or types.messages.ChatFull):
|
|
users = {i.id: i for i in members.users}
|
|
parsed_members = []
|
|
|
|
if isinstance(members, types.channels.ChannelParticipants):
|
|
count = members.count
|
|
members = members.participants
|
|
|
|
for member in members:
|
|
user = parse_user(users[member.user_id])
|
|
|
|
if isinstance(member, (types.ChannelParticipant, types.ChannelParticipantSelf)):
|
|
parsed_members.append(
|
|
pyrogram_types.ChatMember(
|
|
user=user,
|
|
status="member"
|
|
)
|
|
)
|
|
elif isinstance(member, types.ChannelParticipantCreator):
|
|
parsed_members.append(
|
|
pyrogram_types.ChatMember(
|
|
user=user,
|
|
status="creator"
|
|
)
|
|
)
|
|
elif isinstance(member, types.ChannelParticipantAdmin):
|
|
rights = member.admin_rights # type: types.ChannelAdminRights
|
|
|
|
parsed_members.append(
|
|
pyrogram_types.ChatMember(
|
|
user=user,
|
|
status="administrator",
|
|
can_be_edited=member.can_edit,
|
|
can_change_info=rights.change_info,
|
|
can_post_messages=rights.post_messages,
|
|
can_edit_messages=rights.edit_messages,
|
|
can_delete_messages=rights.delete_messages,
|
|
can_invite_users=rights.invite_users or rights.invite_link,
|
|
can_restrict_members=rights.ban_users,
|
|
can_pin_messages=rights.pin_messages,
|
|
can_promote_members=rights.add_admins
|
|
)
|
|
)
|
|
elif isinstance(member, types.ChannelParticipantBanned):
|
|
rights = member.banned_rights # type: types.ChannelBannedRights
|
|
|
|
chat_member = pyrogram_types.ChatMember(
|
|
user=user,
|
|
status="kicked" if rights.view_messages else "restricted",
|
|
until_date=0 if rights.until_date == (1 << 31) - 1 else rights.until_date
|
|
)
|
|
|
|
if chat_member.status == "restricted":
|
|
chat_member.can_send_messages = not rights.send_messages
|
|
chat_member.can_send_media_messages = not rights.send_media
|
|
chat_member.can_send_other_messages = (
|
|
not rights.send_stickers or not rights.send_gifs or
|
|
not rights.send_games or not rights.send_inline
|
|
)
|
|
chat_member.can_add_web_page_previews = not rights.embed_links
|
|
|
|
parsed_members.append(chat_member)
|
|
|
|
return pyrogram_types.ChatMembers(
|
|
total_count=count,
|
|
chat_members=parsed_members
|
|
)
|
|
else:
|
|
members = members.full_chat.participants.participants
|
|
|
|
for member in members:
|
|
user = parse_user(users[member.user_id])
|
|
|
|
if isinstance(member, types.ChatParticipant):
|
|
parsed_members.append(
|
|
pyrogram_types.ChatMember(
|
|
user=user,
|
|
status="member"
|
|
)
|
|
)
|
|
elif isinstance(member, types.ChatParticipantCreator):
|
|
parsed_members.append(
|
|
pyrogram_types.ChatMember(
|
|
user=user,
|
|
status="creator"
|
|
)
|
|
)
|
|
elif isinstance(member, types.ChatParticipantAdmin):
|
|
parsed_members.append(
|
|
pyrogram_types.ChatMember(
|
|
user=user,
|
|
status="administrator"
|
|
)
|
|
)
|
|
|
|
return pyrogram_types.ChatMembers(
|
|
total_count=len(members),
|
|
chat_members=parsed_members
|
|
)
|