Telethon/telethon/utils.py
2018-06-10 11:33:20 +02:00

694 lines
22 KiB
Python

"""
Utilities for working with the Telegram API itself (such as handy methods
to convert between an entity like an User, Chat, etc. into its Input version)
"""
import itertools
import math
import mimetypes
import os
import re
import struct
import types
from collections import UserList
from mimetypes import guess_extension
from .extensions import markdown, html
from .tl import TLObject
from .tl.types import (
Channel, ChannelForbidden, Chat, ChatEmpty, ChatForbidden, ChatFull,
ChatPhoto, InputPeerChannel, InputPeerChat, InputPeerUser, InputPeerEmpty,
MessageMediaDocument, MessageMediaPhoto, PeerChannel, InputChannel,
UserEmpty, InputUser, InputUserEmpty, InputUserSelf, InputPeerSelf,
PeerChat, PeerUser, User, UserFull, UserProfilePhoto, Document,
MessageMediaContact, MessageMediaEmpty, MessageMediaGame, MessageMediaGeo,
MessageMediaUnsupported, MessageMediaVenue, InputMediaContact,
InputMediaDocument, InputMediaEmpty, InputMediaGame,
InputMediaGeoPoint, InputMediaPhoto, InputMediaVenue, InputDocument,
DocumentEmpty, InputDocumentEmpty, Message, GeoPoint, InputGeoPoint,
GeoPointEmpty, InputGeoPointEmpty, Photo, InputPhoto, PhotoEmpty,
InputPhotoEmpty, FileLocation, ChatPhotoEmpty, UserProfilePhotoEmpty,
FileLocationUnavailable, InputMediaUploadedDocument, ChannelFull,
InputMediaUploadedPhoto, DocumentAttributeFilename, photos,
TopPeer, InputNotifyPeer, InputMessageID, InputFileLocation,
InputDocumentFileLocation, PhotoSizeEmpty, InputDialogPeer
)
from .tl.types.contacts import ResolvedPeer
USERNAME_RE = re.compile(
r'@|(?:https?://)?(?:www\.)?(?:telegram\.(?:me|dog)|t\.me)/(joinchat/)?'
)
# The only shorter-than-five-characters usernames are those used for some
# special, very well known bots. This list may be incomplete though:
# "[...] @gif, @vid, @pic, @bing, @wiki, @imdb and @bold [...]"
#
# See https://telegram.org/blog/inline-bots#how-does-it-work
VALID_USERNAME_RE = re.compile(
r'^([a-z][\w\d]{3,30}[a-z\d]'
r'|gif|vid|pic|bing|wiki|imdb|bold|vote|like|coub|ya)$',
re.IGNORECASE
)
class Default:
"""
Sentinel value to indicate that the default value should be used.
Currently used for the ``parse_mode``, where a ``None`` mode should
be considered different from using the default.
"""
def chunks(iterable, size=100):
"""
Turns the given iterable into chunks of the specified size,
which is 100 by default since that's what Telegram uses the most.
"""
it = iter(iterable)
size -= 1
for head in it:
yield itertools.chain([head], itertools.islice(it, size))
def get_display_name(entity):
"""
Gets the display name for the given entity, if it's an :tl:`User`,
:tl:`Chat` or :tl:`Channel`. Returns an empty string otherwise.
"""
if isinstance(entity, User):
if entity.last_name and entity.first_name:
return '{} {}'.format(entity.first_name, entity.last_name)
elif entity.first_name:
return entity.first_name
elif entity.last_name:
return entity.last_name
else:
return ''
elif isinstance(entity, (Chat, Channel)):
return entity.title
return ''
def get_extension(media):
"""Gets the corresponding extension for any Telegram media."""
# Photos are always compressed as .jpg by Telegram
if isinstance(media, (UserProfilePhoto, ChatPhoto, MessageMediaPhoto)):
return '.jpg'
# Documents will come with a mime type
if isinstance(media, MessageMediaDocument):
media = media.document
if isinstance(media, Document):
if media.mime_type == 'application/octet-stream':
# Octet stream are just bytes, which have no default extension
return ''
else:
return guess_extension(media.mime_type) or ''
return ''
def _raise_cast_fail(entity, target):
raise TypeError('Cannot cast {} to any kind of {}.'.format(
type(entity).__name__, target))
def get_input_peer(entity, allow_self=True):
"""
Gets the input peer for the given "entity" (user, chat or channel).
A ``TypeError`` is raised if the given entity isn't a supported type.
"""
try:
if entity.SUBCLASS_OF_ID == 0xc91c90b6: # crc32(b'InputPeer')
return entity
except AttributeError:
# e.g. custom.Dialog (can't cyclic import).
if allow_self and hasattr(entity, 'input_entity'):
return entity.input_entity
elif hasattr(entity, 'entity'):
return get_input_peer(entity.entity)
else:
_raise_cast_fail(entity, 'InputPeer')
if isinstance(entity, User):
if entity.is_self and allow_self:
return InputPeerSelf()
else:
return InputPeerUser(entity.id, entity.access_hash or 0)
if isinstance(entity, (Chat, ChatEmpty, ChatForbidden)):
return InputPeerChat(entity.id)
if isinstance(entity, (Channel, ChannelForbidden)):
return InputPeerChannel(entity.id, entity.access_hash or 0)
if isinstance(entity, InputUser):
return InputPeerUser(entity.user_id, entity.access_hash)
if isinstance(entity, InputChannel):
return InputPeerChannel(entity.channel_id, entity.access_hash)
if isinstance(entity, InputUserSelf):
return InputPeerSelf()
if isinstance(entity, UserEmpty):
return InputPeerEmpty()
if isinstance(entity, UserFull):
return get_input_peer(entity.user)
if isinstance(entity, ChatFull):
return InputPeerChat(entity.id)
if isinstance(entity, PeerChat):
return InputPeerChat(entity.chat_id)
_raise_cast_fail(entity, 'InputPeer')
def get_input_channel(entity):
"""Similar to :meth:`get_input_peer`, but for :tl:`InputChannel`'s alone."""
try:
if entity.SUBCLASS_OF_ID == 0x40f202fd: # crc32(b'InputChannel')
return entity
except AttributeError:
_raise_cast_fail(entity, 'InputChannel')
if isinstance(entity, (Channel, ChannelForbidden)):
return InputChannel(entity.id, entity.access_hash or 0)
if isinstance(entity, InputPeerChannel):
return InputChannel(entity.channel_id, entity.access_hash)
_raise_cast_fail(entity, 'InputChannel')
def get_input_user(entity):
"""Similar to :meth:`get_input_peer`, but for :tl:`InputUser`'s alone."""
try:
if entity.SUBCLASS_OF_ID == 0xe669bf46: # crc32(b'InputUser'):
return entity
except AttributeError:
_raise_cast_fail(entity, 'InputUser')
if isinstance(entity, User):
if entity.is_self:
return InputUserSelf()
else:
return InputUser(entity.id, entity.access_hash or 0)
if isinstance(entity, InputPeerSelf):
return InputUserSelf()
if isinstance(entity, (UserEmpty, InputPeerEmpty)):
return InputUserEmpty()
if isinstance(entity, UserFull):
return get_input_user(entity.user)
if isinstance(entity, InputPeerUser):
return InputUser(entity.user_id, entity.access_hash)
_raise_cast_fail(entity, 'InputUser')
def get_input_dialog(dialog):
"""Similar to :meth:`get_input_peer`, but for dialogs"""
try:
if dialog.SUBCLASS_OF_ID == 0xa21c9795: # crc32(b'InputDialogPeer')
return dialog
if dialog.SUBCLASS_OF_ID == 0xc91c90b6: # crc32(b'InputPeer')
return InputDialogPeer(dialog)
except AttributeError:
_raise_cast_fail(dialog, 'InputDialogPeer')
try:
return InputDialogPeer(get_input_peer(dialog))
except TypeError:
pass
_raise_cast_fail(dialog, 'InputDialogPeer')
def get_input_document(document):
"""Similar to :meth:`get_input_peer`, but for documents"""
try:
if document.SUBCLASS_OF_ID == 0xf33fdb68: # crc32(b'InputDocument'):
return document
except AttributeError:
_raise_cast_fail(document, 'InputDocument')
if isinstance(document, Document):
return InputDocument(id=document.id, access_hash=document.access_hash)
if isinstance(document, DocumentEmpty):
return InputDocumentEmpty()
if isinstance(document, MessageMediaDocument):
return get_input_document(document.document)
if isinstance(document, Message):
return get_input_document(document.media)
_raise_cast_fail(document, 'InputDocument')
def get_input_photo(photo):
"""Similar to :meth:`get_input_peer`, but for photos"""
try:
if photo.SUBCLASS_OF_ID == 0x846363e0: # crc32(b'InputPhoto'):
return photo
except AttributeError:
_raise_cast_fail(photo, 'InputPhoto')
if isinstance(photo, photos.Photo):
photo = photo.photo
if isinstance(photo, Photo):
return InputPhoto(id=photo.id, access_hash=photo.access_hash)
if isinstance(photo, PhotoEmpty):
return InputPhotoEmpty()
_raise_cast_fail(photo, 'InputPhoto')
def get_input_geo(geo):
"""Similar to :meth:`get_input_peer`, but for geo points"""
try:
if geo.SUBCLASS_OF_ID == 0x430d225: # crc32(b'InputGeoPoint'):
return geo
except AttributeError:
_raise_cast_fail(geo, 'InputGeoPoint')
if isinstance(geo, GeoPoint):
return InputGeoPoint(lat=geo.lat, long=geo.long)
if isinstance(geo, GeoPointEmpty):
return InputGeoPointEmpty()
if isinstance(geo, MessageMediaGeo):
return get_input_geo(geo.geo)
if isinstance(geo, Message):
return get_input_geo(geo.media)
_raise_cast_fail(geo, 'InputGeoPoint')
def get_input_media(media, is_photo=False):
"""
Similar to :meth:`get_input_peer`, but for media.
If the media is a file location and ``is_photo`` is known to be ``True``,
it will be treated as an :tl:`InputMediaUploadedPhoto`.
"""
try:
if media.SUBCLASS_OF_ID == 0xfaf846f4: # crc32(b'InputMedia')
return media
elif media.SUBCLASS_OF_ID == 0x846363e0: # crc32(b'InputPhoto')
return InputMediaPhoto(media)
elif media.SUBCLASS_OF_ID == 0xf33fdb68: # crc32(b'InputDocument')
return InputMediaDocument(media)
except AttributeError:
_raise_cast_fail(media, 'InputMedia')
if isinstance(media, MessageMediaPhoto):
return InputMediaPhoto(
id=get_input_photo(media.photo),
ttl_seconds=media.ttl_seconds
)
if isinstance(media, (Photo, photos.Photo, PhotoEmpty)):
return InputMediaPhoto(
id=get_input_photo(media)
)
if isinstance(media, MessageMediaDocument):
return InputMediaDocument(
id=get_input_document(media.document),
ttl_seconds=media.ttl_seconds
)
if isinstance(media, (Document, DocumentEmpty)):
return InputMediaDocument(
id=get_input_document(media)
)
if isinstance(media, FileLocation):
if is_photo:
return InputMediaUploadedPhoto(file=media)
else:
return InputMediaUploadedDocument(
file=media,
mime_type='application/octet-stream', # unknown, assume bytes
attributes=[DocumentAttributeFilename('unnamed')]
)
if isinstance(media, MessageMediaGame):
return InputMediaGame(id=media.game.id)
if isinstance(media, (ChatPhoto, UserProfilePhoto)):
if isinstance(media.photo_big, FileLocationUnavailable):
media = media.photo_small
else:
media = media.photo_big
return get_input_media(media, is_photo=True)
if isinstance(media, MessageMediaContact):
return InputMediaContact(
phone_number=media.phone_number,
first_name=media.first_name,
last_name=media.last_name
)
if isinstance(media, MessageMediaGeo):
return InputMediaGeoPoint(geo_point=get_input_geo(media.geo))
if isinstance(media, MessageMediaVenue):
return InputMediaVenue(
geo_point=get_input_geo(media.geo),
title=media.title,
address=media.address,
provider=media.provider,
venue_id=media.venue_id,
venue_type=''
)
if isinstance(media, (
MessageMediaEmpty, MessageMediaUnsupported,
ChatPhotoEmpty, UserProfilePhotoEmpty, FileLocationUnavailable)):
return InputMediaEmpty()
if isinstance(media, Message):
return get_input_media(media.media, is_photo=is_photo)
_raise_cast_fail(media, 'InputMedia')
def get_input_message(message):
"""Similar to :meth:`get_input_peer`, but for input messages."""
try:
if isinstance(message, int): # This case is really common too
return InputMessageID(message)
elif message.SUBCLASS_OF_ID == 0x54b6bcc5: # crc32(b'InputMessage'):
return message
elif message.SUBCLASS_OF_ID == 0x790009e3: # crc32(b'Message'):
return InputMessageID(message.id)
except AttributeError:
pass
_raise_cast_fail(message, 'InputMedia')
def get_message_id(message):
"""Sanitizes the 'reply_to' parameter a user may send"""
if message is None:
return None
if isinstance(message, int):
return message
if hasattr(message, 'original_message'):
return message.original_message.id
try:
if message.SUBCLASS_OF_ID == 0x790009e3:
# hex(crc32(b'Message')) = 0x790009e3
return message.id
except AttributeError:
pass
raise TypeError('Invalid message type: {}'.format(type(message)))
def sanitize_parse_mode(mode):
"""
Converts the given parse mode into an object with
``parse`` and ``unparse`` callable properties.
"""
if not mode:
return None
if callable(mode):
class CustomMode:
@staticmethod
def unparse(text, entities):
raise NotImplementedError
CustomMode.parse = mode
return CustomMode
elif (all(hasattr(mode, x) for x in ('parse', 'unparse'))
and all(callable(x) for x in (mode.parse, mode.unparse))):
return mode
elif isinstance(mode, str):
try:
return {
'md': markdown,
'markdown': markdown,
'htm': html,
'html': html
}[mode.lower()]
except KeyError:
raise ValueError('Unknown parse mode {}'.format(mode))
else:
raise TypeError('Invalid parse mode type {}'.format(mode))
def get_input_location(location):
"""Similar to :meth:`get_input_peer`, but for input messages."""
try:
if location.SUBCLASS_OF_ID == 0x1523d462:
return location # crc32(b'InputFileLocation'):
except AttributeError:
_raise_cast_fail(location, 'InputFileLocation')
if isinstance(location, Message):
location = location.media
if isinstance(location, MessageMediaDocument):
location = location.document
elif isinstance(location, MessageMediaPhoto):
location = location.photo
if isinstance(location, Document):
return InputDocumentFileLocation(
location.id, location.access_hash, location.version)
elif isinstance(location, Photo):
try:
location = next(x for x in reversed(location.sizes)
if not isinstance(x, PhotoSizeEmpty)).location
except StopIteration:
pass
if isinstance(location, (FileLocation, FileLocationUnavailable)):
return InputFileLocation(
location.volume_id, location.local_id, location.secret)
_raise_cast_fail(location, 'InputFileLocation')
def is_image(file):
"""
Returns ``True`` if the file extension looks like an image file to Telegram.
"""
if not isinstance(file, str):
return False
_, ext = os.path.splitext(file)
return re.match(r'\.(png|jpe?g)', ext, re.IGNORECASE)
def is_audio(file):
"""Returns ``True`` if the file extension looks like an audio file."""
return (isinstance(file, str) and
(mimetypes.guess_type(file)[0] or '').startswith('audio/'))
def is_video(file):
"""Returns ``True`` if the file extension looks like a video file."""
return (isinstance(file, str) and
(mimetypes.guess_type(file)[0] or '').startswith('video/'))
def is_list_like(obj):
"""
Returns ``True`` if the given object looks like a list.
Checking ``if hasattr(obj, '__iter__')`` and ignoring ``str/bytes`` is not
enough. Things like ``open()`` are also iterable (and probably many
other things), so just support the commonly known list-like objects.
"""
return isinstance(obj, (list, tuple, set, dict,
UserList, types.GeneratorType))
def parse_phone(phone):
"""Parses the given phone, or returns ``None`` if it's invalid."""
if isinstance(phone, int):
return str(phone)
else:
phone = re.sub(r'[+()\s-]', '', str(phone))
if phone.isdigit():
return phone
def parse_username(username):
"""Parses the given username or channel access hash, given
a string, username or URL. Returns a tuple consisting of
both the stripped, lowercase username and whether it is
a joinchat/ hash (in which case is not lowercase'd).
Returns ``None`` if the ``username`` is not valid.
"""
username = username.strip()
m = USERNAME_RE.match(username)
if m:
username = username[m.end():]
is_invite = bool(m.group(1))
if is_invite:
return username, True
else:
username = username.rstrip('/')
if VALID_USERNAME_RE.match(username):
return username.lower(), False
else:
return None, False
def _fix_peer_id(peer_id):
"""
Fixes the peer ID for chats and channels, in case the users
mix marking the ID with the :tl:`Peer` constructors.
"""
peer_id = abs(peer_id)
if str(peer_id).startswith('100'):
peer_id = str(peer_id)[3:]
return int(peer_id)
def add_surrogate(text):
return ''.join(
# SMP -> Surrogate Pairs (Telegram offsets are calculated with these).
# See https://en.wikipedia.org/wiki/Plane_(Unicode)#Overview for more.
''.join(chr(y) for y in struct.unpack('<HH', x.encode('utf-16le')))
if (0x10000 <= ord(x) <= 0x10FFFF) else x for x in text
)
def del_surrogate(text):
return text.encode('utf-16', 'surrogatepass').decode('utf-16')
def get_inner_text(text, entity):
"""
Gets the inner text that's surrounded by the given entity or entities.
For instance: text = 'hey!', entity = MessageEntityBold(2, 2) -> 'y!'.
:param text: the original text.
:param entity: the entity or entities that must be matched.
:return: a single result or a list of the text surrounded by the entities.
"""
if isinstance(entity, TLObject):
entity = (entity,)
multiple = True
else:
multiple = False
text = add_surrogate(text)
result = []
for e in entity:
start = e.offset
end = e.offset + e.length
result.append(del_surrogate(text[start:end]))
return result if multiple else result[0]
def get_peer_id(peer):
"""
Finds the ID of the given peer, and converts it to the "bot api" format
so it the peer can be identified back. User ID is left unmodified,
chat ID is negated, and channel ID is prefixed with -100.
The original ID and the peer type class can be returned with
a call to :meth:`resolve_id(marked_id)`.
"""
# First we assert it's a Peer TLObject, or early return for integers
if isinstance(peer, int):
return peer
try:
if peer.SUBCLASS_OF_ID not in (0x2d45687, 0xc91c90b6):
if isinstance(peer, (ResolvedPeer, InputNotifyPeer, TopPeer)):
peer = peer.peer
else:
# Not a Peer or an InputPeer, so first get its Input version
peer = get_input_peer(peer, allow_self=False)
except AttributeError:
_raise_cast_fail(peer, 'int')
# Set the right ID/kind, or raise if the TLObject is not recognised
if isinstance(peer, (PeerUser, InputPeerUser)):
return peer.user_id
elif isinstance(peer, (PeerChat, InputPeerChat)):
# Check in case the user mixed things up to avoid blowing up
if not (0 < peer.chat_id <= 0x7fffffff):
peer.chat_id = _fix_peer_id(peer.chat_id)
return -peer.chat_id
elif isinstance(peer, (PeerChannel, InputPeerChannel, ChannelFull)):
if isinstance(peer, ChannelFull):
# Special case: .get_input_peer can't return InputChannel from
# ChannelFull since it doesn't have an .access_hash attribute.
i = peer.id
else:
i = peer.channel_id
# Check in case the user mixed things up to avoid blowing up
if not (0 < i <= 0x7fffffff):
i = _fix_peer_id(i)
if isinstance(peer, ChannelFull):
peer.id = i
else:
peer.channel_id = i
# Concat -100 through math tricks, .to_supergroup() on Madeline
# IDs will be strictly positive -> log works
return -(i + pow(10, math.floor(math.log10(i) + 3)))
_raise_cast_fail(peer, 'int')
def resolve_id(marked_id):
"""Given a marked ID, returns the original ID and its :tl:`Peer` type."""
if marked_id >= 0:
return marked_id, PeerUser
# There have been report of chat IDs being 10000xyz, which means their
# marked version is -10000xyz, which in turn looks like a channel but
# it becomes 00xyz (= xyz). Hence, we must assert that there are only
# two zeroes.
m = re.match(r'-100([^0]\d*)', str(marked_id))
if m:
return int(m.group(1)), PeerChannel
return -marked_id, PeerChat
def get_appropriated_part_size(file_size):
"""
Gets the appropriated part size when uploading or downloading files,
given an initial file size.
"""
if file_size <= 104857600: # 100MB
return 128
if file_size <= 786432000: # 750MB
return 256
if file_size <= 1572864000: # 1500MB
return 512
raise ValueError('File size too large')