Telethon/telethon/utils/tl_utils.py

90 lines
2.9 KiB
Python

"""Utilities for working with TLObjects.
We have these because some TLObjects differ in very little things,
for example, some may have an `user_id` attribute and other a `chat_id` but,
after all, both are the same attribute, IDs."""
from mimetypes import guess_extension
from telethon.tl.types import \
User, Chat, Channel, \
PeerUser, PeerChat, PeerChannel, \
InputPeerUser, InputPeerChat, InputPeerChannel, \
UserProfilePhoto, ChatPhoto, \
MessageMediaPhoto, MessageMediaDocument
def get_display_name(entity):
"""Gets the input peer for the given "entity" (user, chat or channel)
Returns None if it was not found"""
if isinstance(entity, User):
if entity.last_name is not None:
return '{} {}'.format(entity.first_name, entity.last_name)
return entity.first_name
if isinstance(entity, Chat) or isinstance(entity, Channel):
return entity.title
def get_extension(media):
"""Gets the corresponding extension for any Telegram media"""
# Photos are always compressed as .jpg by Telegram
if (isinstance(media, UserProfilePhoto) or
isinstance(media, ChatPhoto) or
isinstance(media, MessageMediaPhoto)):
return '.jpg'
# Documents will come with a mime type, from which we can guess their mime type
if isinstance(media, MessageMediaDocument):
extension = guess_extension(media.document.mime_type)
return extension if extension else ''
return None
def get_input_peer(entity):
"""Gets the input peer for the given "entity" (user, chat or channel).
Returns None if it was not found"""
if isinstance(entity, User):
return InputPeerUser(entity.id, entity.access_hash)
if isinstance(entity, Chat):
return InputPeerChat(entity.id)
if isinstance(entity, Channel):
return InputPeerChannel(entity.id, entity.access_hash)
def find_user_or_chat(peer, users, chats):
"""Finds the corresponding user or chat given a peer.
Returns None if it was not found"""
try:
if isinstance(peer, PeerUser):
user = next(u for u in users if u.id == peer.user_id)
return user
elif isinstance(peer, PeerChat):
chat = next(c for c in chats if c.id == peer.chat_id)
return chat
elif isinstance(peer, PeerChannel):
channel = next(c for c in chats if c.id == peer.channel_id)
return channel
except StopIteration:
return None
def get_appropiate_part_size(file_size):
"""Gets the appropiate part size when uploading or downloading files,
given an initial file size"""
if file_size <= 1048576: # 1MB
return 32
if file_size <= 10485760: # 10MB
return 64
if file_size <= 393216000: # 375MB
return 128
if file_size <= 786432000: # 750MB
return 256
if file_size <= 1572864000: # 1500MB
return 512
raise ValueError('File size too large')