diff --git a/readthedocs/custom_roles.py b/readthedocs/custom_roles.py index bf025fb85..6d13ee8d4 100644 --- a/readthedocs/custom_roles.py +++ b/readthedocs/custom_roles.py @@ -22,10 +22,9 @@ def make_link_node(rawtext, app, name, options): base += '/' set_classes(options) - node = nodes.reference(rawtext, utils.unescape(name), - refuri='{}?q={}'.format(base, name), - **options) - return node + return nodes.reference( + rawtext, utils.unescape(name), refuri=f'{base}?q={name}', **options + ) # noinspection PyUnusedLocal diff --git a/setup.py b/setup.py index a498980a6..a0ba7e789 100755 --- a/setup.py +++ b/setup.py @@ -166,10 +166,12 @@ def main(argv): return remove_dirs = ['__pycache__', 'build', 'dist', 'Telethon.egg-info'] - for root, _dirs, _files in os.walk(LIBRARY_DIR, topdown=False): - # setuptools is including __pycache__ for some reason (#1605) - if root.endswith('/__pycache__'): - remove_dirs.append(root) + remove_dirs.extend( + root + for root, _dirs, _files in os.walk(LIBRARY_DIR, topdown=False) + if root.endswith('/__pycache__') + ) + for x in remove_dirs: shutil.rmtree(x, ignore_errors=True) diff --git a/telethon/client/account.py b/telethon/client/account.py index d82235b6a..1b750e508 100644 --- a/telethon/client/account.py +++ b/telethon/client/account.py @@ -101,7 +101,7 @@ def __getattr__(self, name): return value def __setattr__(self, name, value): - if name.startswith('_{}__'.format(type(self).__name__.lstrip('_'))): + if name.startswith(f"_{type(self).__name__.lstrip('_')}__"): # This is our own name-mangled attribute, keep calm. return super().__setattr__(name, value) return setattr(self.__client, name, value) diff --git a/telethon/client/auth.py b/telethon/client/auth.py index 9665262bb..24c0e60ba 100644 --- a/telethon/client/auth.py +++ b/telethon/client/auth.py @@ -222,10 +222,10 @@ async def _start( attempts += 1 else: raise RuntimeError( - '{} consecutive sign-in attempts failed. Aborting' - .format(max_attempts) + f'{max_attempts} consecutive sign-in attempts failed. Aborting' ) + if two_step_detected: if not password: raise ValueError( diff --git a/telethon/client/chats.py b/telethon/client/chats.py index a854447ad..8d1501105 100644 --- a/telethon/client/chats.py +++ b/telethon/client/chats.py @@ -277,8 +277,9 @@ async def _init( if not utils.is_list_like(admins): admins = (admins,) - for admin in admins: - admin_list.append(await self.client.get_input_entity(admin)) + admin_list.extend( + await self.client.get_input_entity(admin) for admin in admins + ) self.request = functions.channels.GetAdminLogRequest( self.entity, q=search or '', min_id=min_id, max_id=max_id, @@ -825,14 +826,13 @@ def action( try: action = _ChatAction._str_mapping[action.lower()] except KeyError: - raise ValueError( - 'No such action "{}"'.format(action)) from None + raise ValueError(f'No such action "{action}"') from None elif not isinstance(action, types.TLObject) or action.SUBCLASS_OF_ID != 0x20b2cc21: # 0x20b2cc21 = crc32(b'SendMessageAction') if isinstance(action, type): raise ValueError('You must pass an instance, not the class') else: - raise ValueError('Cannot use {} as action'.format(action)) + raise ValueError(f'Cannot use {action} as action') if isinstance(action, types.SendMessageCancelAction): # ``SetTypingRequest.resolve`` will get input peer of ``entity``. diff --git a/telethon/client/dialogs.py b/telethon/client/dialogs.py index 8c0860fce..b965952a9 100644 --- a/telethon/client/dialogs.py +++ b/telethon/client/dialogs.py @@ -116,10 +116,8 @@ async def _init(self, entities, **kwargs): items = r.updates else: peers = [] - for entity in entities: - peers.append(types.InputDialogPeer( - await self.client.get_input_entity(entity))) - + peers.extend(types.InputDialogPeer( + await self.client.get_input_entity(entity)) for entity in entities) r = await self.client(functions.messages.GetPeerDialogsRequest(peers)) items = r.dialogs @@ -310,10 +308,7 @@ async def get_drafts( print(drafts.text) """ items = await self.iter_drafts(entity).collect() - if not entity or utils.is_list_like(entity): - return items - else: - return items[0] + return items if not entity or utils.is_list_like(entity) else items[0] async def edit_folder( self: 'TelegramClient', @@ -380,11 +375,11 @@ async def edit_folder( folder_id=unpack )) - if not utils.is_list_like(entity): - entities = [await self.get_input_entity(entity)] - else: - entities = await asyncio.gather( - *(self.get_input_entity(x) for x in entity)) + entities = ( + await asyncio.gather(*(self.get_input_entity(x) for x in entity)) + if utils.is_list_like(entity) + else [await self.get_input_entity(entity)] + ) if folder is None: raise ValueError('You must specify a folder') @@ -443,11 +438,7 @@ async def delete_dialog( """ # If we have enough information (`Dialog.delete` gives it to us), # then we know we don't have to kick ourselves in deactivated chats. - if isinstance(entity, types.Chat): - deactivated = entity.deactivated - else: - deactivated = False - + deactivated = entity.deactivated if isinstance(entity, types.Chat) else False entity = await self.get_input_entity(entity) ty = helpers._entity_type(entity) if ty == helpers._EntityType.CHANNEL: diff --git a/telethon/client/downloads.py b/telethon/client/downloads.py index 6d7a8d658..c1105f726 100644 --- a/telethon/client/downloads.py +++ b/telethon/client/downloads.py @@ -256,33 +256,30 @@ async def download_profile_photo( if not hasattr(entity, 'photo'): # Special case: may be a ChatFull with photo:Photo # This is different from a normal UserProfilePhoto and Chat - if not hasattr(entity, 'chat_photo'): - return None - return await self._download_photo( entity.chat_photo, file, date=None, thumb=thumb, progress_callback=None - ) - - for attr in ('username', 'first_name', 'title'): - possible_names.append(getattr(entity, attr, None)) + ) if hasattr(entity, 'chat_photo') else None + possible_names.extend( + getattr(entity, attr, None) + for attr in ('username', 'first_name', 'title') + ) photo = entity.photo - if isinstance(photo, (types.UserProfilePhoto, types.ChatPhoto)): - dc_id = photo.dc_id - loc = types.InputPeerPhotoFileLocation( - peer=await self.get_input_entity(entity), - photo_id=photo.photo_id, - big=download_big - ) - else: + if not isinstance(photo, (types.UserProfilePhoto, types.ChatPhoto)): # It doesn't make any sense to check if `photo` can be used # as input location, because then this method would be able # to "download the profile photo of a message", i.e. its # media which should be done with `download_media` instead. return None + dc_id = photo.dc_id + loc = types.InputPeerPhotoFileLocation( + peer=await self.get_input_entity(entity), + photo_id=photo.photo_id, + big=download_big + ) file = self._get_proper_filename( file, 'profile_photo', '.jpg', possible_names=possible_names @@ -296,16 +293,15 @@ async def download_profile_photo( # The fix seems to be using the full channel chat photo. ie = await self.get_input_entity(entity) ty = helpers._entity_type(ie) - if ty == helpers._EntityType.CHANNEL: - full = await self(functions.channels.GetFullChannelRequest(ie)) - return await self._download_photo( - full.full_chat.chat_photo, file, - date=None, progress_callback=None, - thumb=thumb - ) - else: + if ty != helpers._EntityType.CHANNEL: # Until there's a report for chats, no need to. return None + full = await self(functions.channels.GetFullChannelRequest(ie)) + return await self._download_photo( + full.full_chat.chat_photo, file, + date=None, progress_callback=None, + thumb=thumb + ) async def download_media( self: 'TelegramClient', @@ -398,14 +394,15 @@ def callback(current, total): if isinstance(media, str): media = utils.resolve_bot_file_id(media) - if isinstance(media, types.MessageService): - if isinstance(message.action, - types.MessageActionChatEditPhoto): - media = media.photo - - if isinstance(media, types.MessageMediaWebPage): - if isinstance(media.webpage, types.WebPage): - media = media.webpage.document or media.webpage.photo + if isinstance(media, types.MessageService) and isinstance( + message.action, types.MessageActionChatEditPhoto + ): + media = media.photo + + if isinstance(media, types.MessageMediaWebPage) and isinstance( + media.webpage, types.WebPage + ): + media = media.webpage.document or media.webpage.photo if isinstance(media, (types.MessageMediaPhoto, types.Photo)): return await self._download_photo( @@ -511,11 +508,7 @@ async def _download_file( iv: bytes = None, msg_data: tuple = None) -> typing.Optional[bytes]: if not part_size_kb: - if not file_size: - part_size_kb = 64 # Reasonable default - else: - part_size_kb = utils.get_appropriated_part_size(file_size) - + part_size_kb = utils.get_appropriated_part_size(file_size) if file_size else 64 part_size = int(part_size_kb * 1024) if part_size % MIN_CHUNK_SIZE != 0: raise ValueError( @@ -846,9 +839,7 @@ def _get_kind_and_names(attributes): elif isinstance(attr, types.DocumentAttributeAudio): kind = 'audio' if attr.performer and attr.title: - possible_names.append('{} - {}'.format( - attr.performer, attr.title - )) + possible_names.append(f'{attr.performer} - {attr.title}') elif attr.performer: possible_names.append(attr.performer) elif attr.title: @@ -1035,7 +1026,7 @@ def _get_proper_filename(file, kind, extension, i = 1 while True: - result = os.path.join(directory, '{} ({}){}'.format(name, i, ext)) + result = os.path.join(directory, f'{name} ({i}){ext}') if not os.path.isfile(result): return result i += 1 diff --git a/telethon/client/messageparse.py b/telethon/client/messageparse.py index 322c541e6..361cbd45a 100644 --- a/telethon/client/messageparse.py +++ b/telethon/client/messageparse.py @@ -87,12 +87,11 @@ async def _parse_message_text(self: 'TelegramClient', message, parse_mode): message, msg_entities = parse_mode.parse(message) if original_message and not message and not msg_entities: raise ValueError("Failed to parse message") - + for i in reversed(range(len(msg_entities))): e = msg_entities[i] if isinstance(e, types.MessageEntityTextUrl): - m = re.match(r'^@|\+|tg://user\?id=(\d+)', e.url) - if m: + if m := re.match(r'^@|\+|tg://user\?id=(\d+)', e.url): user = int(m.group(1)) if m.group(1) else e.url is_mention = await self._replace_with_mention(msg_entities, i, user) if not is_mention: diff --git a/telethon/client/messages.py b/telethon/client/messages.py index 77804a18a..9f53875f2 100644 --- a/telethon/client/messages.py +++ b/telethon/client/messages.py @@ -37,17 +37,15 @@ async def _init( # and simply stopping once we hit a message with ID <= min_id. if self.reverse: offset_id = max(offset_id, min_id) - if offset_id and max_id: - if max_id - offset_id <= 1: - raise StopAsyncIteration + if offset_id and max_id and max_id - offset_id <= 1: + raise StopAsyncIteration if not max_id: max_id = float('inf') else: offset_id = max(offset_id, max_id) - if offset_id and min_id: - if offset_id - min_id <= 1: - raise StopAsyncIteration + if offset_id and min_id and offset_id - min_id <= 1: + raise StopAsyncIteration if self.reverse: if offset_id: @@ -225,14 +223,12 @@ def _message_in_range(self, message): Determine whether the given message is in the range or it should be ignored (and avoid loading more chunks). """ - # No entity means message IDs between chats may vary if self.entity: if self.reverse: if message.id <= self.last_id or message.id >= self.max_id: return False - else: - if message.id >= self.last_id or message.id <= self.min_id: - return False + elif message.id >= self.last_id or message.id <= self.min_id: + return False return True @@ -256,11 +252,7 @@ def _update_offset(self, last_message, response): self.request.offset_date = last_message.date if isinstance(self.request, functions.messages.SearchGlobalRequest): - if last_message.input_chat: - self.request.offset_peer = last_message.input_chat - else: - self.request.offset_peer = types.InputPeerEmpty() - + self.request.offset_peer = last_message.input_chat or types.InputPeerEmpty() self.request.offset_rate = getattr(response, 'next_rate', 0) @@ -568,11 +560,7 @@ async def get_messages(self: 'TelegramClient', *args, **kwargs) -> 'hints.TotalL message_1337 = await client.get_messages(chat, ids=1337) """ if len(args) == 1 and 'limit' not in kwargs: - if 'min_id' in kwargs and 'max_id' in kwargs: - kwargs['limit'] = None - else: - kwargs['limit'] = 1 - + kwargs['limit'] = None if 'min_id' in kwargs and 'max_id' in kwargs else 1 it = self.iter_messages(*args, **kwargs) ids = kwargs.get('ids') @@ -991,7 +979,7 @@ def get_key(m): elif isinstance(m, types.Message): return m.chat_id else: - raise TypeError('Cannot forward messages of type {}'.format(type(m))) + raise TypeError(f'Cannot forward messages of type {type(m)}') sent = [] for _chat_id, chunk in itertools.groupby(messages, key=get_key): @@ -1171,15 +1159,14 @@ async def editmessage( # Invoke `messages.editInlineBotMessage` from the right datacenter. # Otherwise, Telegram will error with `MESSAGE_ID_INVALID` and do nothing. exported = self.session.dc_id != entity.dc_id - if exported: - try: - sender = await self._borrow_exported_sender(entity.dc_id) - return await self._call(sender, request) - finally: - await self._return_exported_sender(sender) - else: + if not exported: return await self(request) + try: + sender = await self._borrow_exported_sender(entity.dc_id) + return await self._call(sender, request) + finally: + await self._return_exported_sender(sender) entity = await self.get_input_entity(entity) request = functions.messages.EditMessageRequest( peer=entity, @@ -1191,8 +1178,7 @@ async def editmessage( reply_markup=self.build_reply_markup(buttons), schedule_date=schedule ) - msg = self._get_response_message(request, await self(request), entity) - return msg + return self._get_response_message(request, await self(request), entity) async def delete_messages( self: 'TelegramClient', @@ -1322,14 +1308,15 @@ async def send_read_acknowledge( await client.send_read_acknowledge(chat, messages) """ if max_id is None: - if not message: - max_id = 0 - else: - if utils.is_list_like(message): - max_id = max(msg.id for msg in message) - else: - max_id = message.id + if message: + max_id = ( + max(msg.id for msg in message) + if utils.is_list_like(message) + else message.id + ) + else: + max_id = 0 entity = await self.get_input_entity(entity) if clear_mentions: await self(functions.messages.ReadMentionsRequest(entity)) diff --git a/telethon/client/telegrambaseclient.py b/telethon/client/telegrambaseclient.py index 494daf9cc..d213c0a2f 100644 --- a/telethon/client/telegrambaseclient.py +++ b/telethon/client/telegrambaseclient.py @@ -575,16 +575,15 @@ def disconnect(self: 'TelegramClient'): """ if self.loop.is_running(): return self._disconnect_coro() - else: - try: - self.loop.run_until_complete(self._disconnect_coro()) - except RuntimeError: - # Python 3.5.x complains when called from - # `__aexit__` and there were pending updates with: - # "Event loop stopped before Future completed." - # - # However, it doesn't really make a lot of sense. - pass + try: + self.loop.run_until_complete(self._disconnect_coro()) + except RuntimeError: + # Python 3.5.x complains when called from + # `__aexit__` and there were pending updates with: + # "Event loop stopped before Future completed." + # + # However, it doesn't really make a lot of sense. + pass def set_proxy(self: 'TelegramClient', proxy: typing.Union[tuple, dict]): """ @@ -596,19 +595,17 @@ def set_proxy(self: 'TelegramClient', proxy: typing.Union[tuple, dict]): - on a call `await client.connect()` (after complete disconnect) - on auto-reconnect attempt (e.g, after previous connection was lost) """ - init_proxy = None if not issubclass(self._connection, TcpMTProxy) else \ + init_proxy = ( types.InputClientProxy(*self._connection.address_info(proxy)) + if issubclass(self._connection, TcpMTProxy) + else None + ) + self._init_request.proxy = init_proxy self._proxy = proxy - # While `await client.connect()` passes new proxy on each new call, - # auto-reconnect attempts use already set up `_connection` inside - # the `_sender`, so the only way to change proxy between those - # is to directly inject parameters. - - connection = getattr(self._sender, "_connection", None) - if connection: + if connection := getattr(self._sender, "_connection", None): if isinstance(connection, TcpMTProxy): connection._ip = proxy[0] connection._port = proxy[1] @@ -808,27 +805,6 @@ async def _get_cdn_client(self: 'TelegramClient', cdn_redirect): """Similar to ._borrow_exported_client, but for CDNs""" # TODO Implement raise NotImplementedError - session = self._exported_sessions.get(cdn_redirect.dc_id) - if not session: - dc = await self._get_dc(cdn_redirect.dc_id, cdn=True) - session = self.session.clone() - await session.set_dc(dc.id, dc.ip_address, dc.port) - self._exported_sessions[cdn_redirect.dc_id] = session - - self._log[__name__].info('Creating new CDN client') - client = TelegramBaseClient( - session, self.api_id, self.api_hash, - proxy=self._sender.connection.conn.proxy, - timeout=self._sender.connection.get_timeout() - ) - - # This will make use of the new RSA keys for this specific CDN. - # - # We won't be calling GetConfigRequest because it's only called - # when needed by ._get_dc, and also it's static so it's likely - # set already. Avoid invoking non-CDN methods by not syncing updates. - client.connect(_sync_updates=False) - return client # endregion diff --git a/telethon/client/updates.py b/telethon/client/updates.py index bcc983f35..b5698de22 100644 --- a/telethon/client/updates.py +++ b/telethon/client/updates.py @@ -397,27 +397,25 @@ async def _dispatch_queue_updates(self: 'TelegramClient'): self._dispatching_updates_queue.clear() async def _dispatch_update(self: 'TelegramClient', update, others, channel_id, pts_date): - if not self._entity_cache.ensure_cached(update): - # We could add a lock to not fetch the same pts twice if we are - # already fetching it. However this does not happen in practice, - # which makes sense, because different updates have different pts. - if self._state_cache.update(update, check_only=True): - # If the update doesn't have pts, fetching won't do anything. - # For example, UpdateUserStatus or UpdateChatUserTyping. - try: - await self._get_difference(update, channel_id, pts_date) - except OSError: - pass # We were disconnected, that's okay - except errors.RPCError: - # There's a high chance the request fails because we lack - # the channel. Because these "happen sporadically" (#1428) - # we should be okay (no flood waits) even if more occur. - pass - except ValueError: - # There is a chance that GetFullChannelRequest and GetDifferenceRequest - # inside the _get_difference() function will end up with - # ValueError("Request was unsuccessful N time(s)") for whatever reasons. - pass + if not self._entity_cache.ensure_cached( + update + ) and self._state_cache.update(update, check_only=True): + # If the update doesn't have pts, fetching won't do anything. + # For example, UpdateUserStatus or UpdateChatUserTyping. + try: + await self._get_difference(update, channel_id, pts_date) + except OSError: + pass # We were disconnected, that's okay + except errors.RPCError: + # There's a high chance the request fails because we lack + # the channel. Because these "happen sporadically" (#1428) + # we should be okay (no flood waits) even if more occur. + pass + except ValueError: + # There is a chance that GetFullChannelRequest and GetDifferenceRequest + # inside the _get_difference() function will end up with + # ValueError("Request was unsuccessful N time(s)") for whatever reasons. + pass if not self._self_input_peer: # Some updates require our own ID, so we must make sure @@ -434,16 +432,13 @@ async def _dispatch_update(self: 'TelegramClient', update, others, channel_id, p built = EventBuilderDict(self, update, others) for conv_set in self._conversations.values(): for conv in conv_set: - ev = built[events.NewMessage] - if ev: + if ev := built[events.NewMessage]: conv._on_new_message(ev) - ev = built[events.MessageEdited] - if ev: + if ev := built[events.MessageEdited]: conv._on_edit(ev) - ev = built[events.MessageRead] - if ev: + if ev := built[events.MessageRead]: conv._on_read(ev) if conv._custom: @@ -538,7 +533,10 @@ async def _get_difference(self: 'TelegramClient', update, channel_id, pts_date): # There are reports where we somehow call get channel difference # with `InputPeerEmpty`. Check our assumptions to better debug # this when it happens. - assert isinstance(channel_id, int), 'channel_id was {}, not int in {}'.format(type(channel_id), update) + assert isinstance( + channel_id, int + ), f'channel_id was {type(channel_id)}, not int in {update}' + try: # Wrap the ID inside a peer to ensure we get a channel back. where = await self.get_input_entity(types.PeerChannel(channel_id)) @@ -595,37 +593,6 @@ async def _handle_auto_reconnect(self: 'TelegramClient'): 'after reconnect: %s: %s', type(e), e) return - try: - self._log[__name__].info( - 'Asking for the current state after reconnect...') - - # TODO consider: - # If there aren't many updates while the client is disconnected - # (I tried with up to 20), Telegram seems to send them without - # asking for them (via updates.getDifference). - # - # On disconnection, the library should probably set a "need - # difference" or "catching up" flag so that any new updates are - # ignored, and then the library should call updates.getDifference - # itself to fetch them. - # - # In any case (either there are too many updates and Telegram - # didn't send them, or there isn't a lot and Telegram sent them - # but we dropped them), we fetch the new difference to get all - # missed updates. I feel like this would be the best solution. - - # If a disconnection occurs, the old known state will be - # the latest one we were aware of, so we can catch up since - # the most recent state we were aware of. - await self.catch_up() - - self._log[__name__].info('Successfully fetched missed updates') - except errors.RPCError as e: - self._log[__name__].warning('Failed to get missed updates after ' - 'reconnect: %r', e) - except Exception: - self._log[__name__].exception( - 'Unhandled exception while getting update difference after reconnect') # endregion diff --git a/telethon/client/uploads.py b/telethon/client/uploads.py index f2dd1ca74..6b64da836 100644 --- a/telethon/client/uploads.py +++ b/telethon/client/uploads.py @@ -352,11 +352,7 @@ def callback(current, total): # First check if the user passed an iterable, in which case # we may want to send grouped. if utils.is_list_like(file): - if utils.is_list_like(caption): - captions = caption - else: - captions = [caption] - + captions = caption if utils.is_list_like(caption) else [caption] result = [] while file: result += await self._send_album( @@ -431,8 +427,10 @@ async def _send_album(self: 'TelegramClient', entity, files, caption='', caption = (caption,) captions = [] - for c in reversed(caption): # Pop from the end (so reverse) - captions.append(await self._parse_message_text(c or '', parse_mode)) + captions.extend( + await self._parse_message_text(c or '', parse_mode) + for c in reversed(caption) + ) reply_to = utils.get_message_id(reply_to) @@ -460,10 +458,7 @@ async def _send_album(self: 'TelegramClient', entity, files, caption='', fm = utils.get_input_media( r.document, supports_streaming=supports_streaming) - if captions: - caption, msg_entities = captions.pop() - else: - caption, msg_entities = '', None + caption, msg_entities = captions.pop() if captions else ('', None) media.append(types.InputSingleMedia( fm, message=caption, @@ -613,15 +608,17 @@ async def upload_file( if not isinstance(part, bytes): raise TypeError( - 'file descriptor returned {}, not bytes (you must ' - 'open the file in bytes mode)'.format(type(part))) + f'file descriptor returned {type(part)}, not bytes (you must open the file in bytes mode)' + ) + # `file_size` could be wrong in which case `part` may not be # `part_size` before reaching the end. if len(part) != part_size and part_index < part_count - 1: raise ValueError( - 'read less than {} before reaching the end; either ' - '`file_size` or `read` are wrong'.format(part_size)) + f'read less than {part_size} before reaching the end; either `file_size` or `read` are wrong' + ) + pos += len(part) @@ -645,15 +642,13 @@ async def upload_file( file_id, part_index, part) result = await self(request) - if result: - self._log[__name__].debug('Uploaded %d/%d', - part_index + 1, part_count) - if progress_callback: - await helpers._maybe_await(progress_callback(pos, file_size)) - else: - raise RuntimeError( - 'Failed to upload file part {}.'.format(part_index)) + if not result: + raise RuntimeError(f'Failed to upload file part {part_index}.') + self._log[__name__].debug('Uploaded %d/%d', + part_index + 1, part_count) + if progress_callback: + await helpers._maybe_await(progress_callback(pos, file_size)) if is_big: return types.InputFileBig(file_id, part_count, file_name) else: @@ -716,22 +711,22 @@ async def _file_to_media( progress_callback=progress_callback ) elif re.match('https?://', file): - if as_image: - media = types.InputMediaPhotoExternal(file, ttl_seconds=ttl) - else: - media = types.InputMediaDocumentExternal(file, ttl_seconds=ttl) - else: - bot_file = utils.resolve_bot_file_id(file) - if bot_file: - media = utils.get_input_media(bot_file, ttl=ttl) + media = ( + types.InputMediaPhotoExternal(file, ttl_seconds=ttl) + if as_image + else types.InputMediaDocumentExternal(file, ttl_seconds=ttl) + ) + + elif bot_file := utils.resolve_bot_file_id(file): + media = utils.get_input_media(bot_file, ttl=ttl) if media: pass # Already have media, don't check the rest elif not file_handle: raise ValueError( - 'Failed to convert {} to media. Not an existing file, ' - 'an HTTP URL or a valid bot-API-like file ID'.format(file) + f'Failed to convert {file} to media. Not an existing file, an HTTP URL or a valid bot-API-like file ID' ) + elif as_image: media = types.InputMediaUploadedPhoto(file_handle, ttl_seconds=ttl) else: diff --git a/telethon/client/users.py b/telethon/client/users.py index 22db969e3..de2422d58 100644 --- a/telethon/client/users.py +++ b/telethon/client/users.py @@ -91,7 +91,9 @@ async def _call(self: 'TelegramClient', sender, request, ordered=False, flood_sl last_error = e self._log[__name__].warning( 'Telegram is having internal issues %s: %s', - e.__class__.__name__, e) + last_error.__class__.__name__, + last_error, + ) await asyncio.sleep(2) except (errors.FloodWaitError, errors.SlowModeWaitError, errors.FloodTestPhoneWaitError) as e: @@ -109,11 +111,10 @@ async def _call(self: 'TelegramClient', sender, request, ordered=False, flood_sl if e.seconds == 0: e.seconds = 1 - if e.seconds <= self.flood_sleep_threshold: - self._log[__name__].info(*_fmt_flood(e.seconds, request)) - await asyncio.sleep(e.seconds) - else: + if e.seconds > self.flood_sleep_threshold: raise + self._log[__name__].info(*_fmt_flood(e.seconds, request)) + await asyncio.sleep(e.seconds) except (errors.PhoneMigrateError, errors.NetworkMigrateError, errors.UserMigrateError) as e: last_error = e @@ -127,8 +128,7 @@ async def _call(self: 'TelegramClient', sender, request, ordered=False, flood_sl if self._raise_last_call_error and last_error is not None: raise last_error - raise ValueError('Request was unsuccessful {} time(s)' - .format(attempt)) + raise ValueError(f'Request was unsuccessful {attempt} time(s)') # region Public methods @@ -464,10 +464,7 @@ async def get_input_entity( pass raise ValueError( - 'Could not find the input entity for {} ({}). Please read https://' - 'docs.telethon.dev/en/latest/concepts/entities.html to' - ' find out more details.' - .format(peer, type(peer).__name__) + f'Could not find the input entity for {peer} ({type(peer).__name__}). Please read https://docs.telethon.dev/en/latest/concepts/entities.html to find out more details.' ) async def _get_peer(self: 'TelegramClient', peer: 'hints.EntityLike'): @@ -553,8 +550,7 @@ async def _get_entity_from_string(self: 'TelegramClient', string): result = await self( functions.contacts.ResolveUsernameRequest(username)) except errors.UsernameNotOccupiedError as e: - raise ValueError('No user has "{}" as username' - .format(username)) from e + raise ValueError(f'No user has "{username}" as username') from e try: pid = utils.get_peer_id(result.peer, add_mark=False) @@ -571,9 +567,7 @@ async def _get_entity_from_string(self: 'TelegramClient', string): except ValueError: pass - raise ValueError( - 'Cannot find any entity corresponding to "{}"'.format(string) - ) + raise ValueError(f'Cannot find any entity corresponding to "{string}"') async def _get_input_dialog(self: 'TelegramClient', dialog): """ diff --git a/telethon/crypto/aes.py b/telethon/crypto/aes.py index 1a3854334..f6985a1c5 100644 --- a/telethon/crypto/aes.py +++ b/telethon/crypto/aes.py @@ -81,10 +81,9 @@ def encrypt_ige(plain_text, key, iv): Encrypts the given text in 16-bytes blocks by using the given key and 32-bytes initialization vector. """ - padding = len(plain_text) % 16 - if padding: + if padding := len(plain_text) % 16: plain_text += os.urandom(16 - padding) - + if tgcrypto: return tgcrypto.ige256_encrypt(plain_text, key, iv) if cryptg: diff --git a/telethon/crypto/factorization.py b/telethon/crypto/factorization.py index 064756ce6..0f186e1f5 100644 --- a/telethon/crypto/factorization.py +++ b/telethon/crypto/factorization.py @@ -27,13 +27,13 @@ def factorize(cls, pq): while g == 1: x = y - for i in range(r): + for _ in range(r): y = (pow(y, 2, pq) + c) % pq k = 0 while k < r and g == 1: ys = y - for i in range(min(m, r - k)): + for _ in range(min(m, r - k)): y = (pow(y, 2, pq) + c) % pq q = q * (abs(x - y)) % pq diff --git a/telethon/crypto/libssl.py b/telethon/crypto/libssl.py index e0fea00e2..d56e06294 100644 --- a/telethon/crypto/libssl.py +++ b/telethon/crypto/libssl.py @@ -67,8 +67,7 @@ def _find_ssl_lib(): # Fix for https://github.com/LonamiWebs/Telethon/issues/1167 lib = os.path.realpath(os.path.join(root, lib)) return ctypes.cdll.LoadLibrary(lib) - else: - raise OSError('no absolute path for "%s" and cannot load by name' % lib) + raise OSError('no absolute path for "%s" and cannot load by name' % lib) try: diff --git a/telethon/crypto/rsa.py b/telethon/crypto/rsa.py index 91ca7bad5..cc846b2df 100644 --- a/telethon/crypto/rsa.py +++ b/telethon/crypto/rsa.py @@ -77,9 +77,7 @@ def encrypt(fingerprint, data, *, use_old=False): # rsa module uses rsa.transform.bytes2int(to_encrypt), easier way: payload = int.from_bytes(to_encrypt, 'big') encrypted = rsa.core.encrypt_int(payload, key.e, key.n) - # rsa module uses transform.int2bytes(encrypted, keylength), easier: - block = encrypted.to_bytes(256, 'big') - return block + return encrypted.to_bytes(256, 'big') # Add default keys diff --git a/telethon/entitycache.py b/telethon/entitycache.py index b6d06b3c7..a025e757c 100644 --- a/telethon/entitycache.py +++ b/telethon/entitycache.py @@ -85,8 +85,7 @@ def __getitem__(self, item): raise KeyError('Invalid key will not have entity') from None for cls in (types.PeerUser, types.PeerChat, types.PeerChannel): - result = self.__dict__.get(utils.get_peer_id(cls(item))) - if result: + if result := self.__dict__.get(utils.get_peer_id(cls(item))): return result raise KeyError('No cached entity for the given key') diff --git a/telethon/errors/__init__.py b/telethon/errors/__init__.py index f6bc16e5c..70f6273ac 100644 --- a/telethon/errors/__init__.py +++ b/telethon/errors/__init__.py @@ -30,8 +30,7 @@ def rpc_message_to_error(rpc_error, request): return cls(request=request) for msg_regex, cls in rpc_errors_re: - m = re.match(msg_regex, rpc_error.error_message) - if m: + if m := re.match(msg_regex, rpc_error.error_message): capture = int(m.group(1)) if m.groups() else None return cls(request=request, capture=capture) diff --git a/telethon/errors/common.py b/telethon/errors/common.py index de7d95f89..2c125c0dc 100644 --- a/telethon/errors/common.py +++ b/telethon/errors/common.py @@ -33,9 +33,9 @@ class InvalidChecksumError(Exception): """ def __init__(self, checksum, valid_checksum): super().__init__( - 'Invalid checksum ({} when {} was expected). ' - 'This packet should be skipped.' - .format(checksum, valid_checksum)) + f'Invalid checksum ({checksum} when {valid_checksum} was expected). This packet should be skipped.' + ) + self.checksum = checksum self.valid_checksum = valid_checksum @@ -50,12 +50,10 @@ def __init__(self, payload): self.payload = payload if len(payload) == 4: self.code = -struct.unpack('= 0) and (len(result) != length): raise BufferError( - 'No more data left to read (need {}, got {}: {}); last read {}' - .format(length, len(result), repr(result), repr(self._last)) + f'No more data left to read (need {length}, got {len(result)}: {repr(result)}); last read {repr(self._last)}' ) + self._last = result return result @@ -106,7 +106,7 @@ def tgread_bool(self): elif value == 0xbc799737: # boolFalse return False else: - raise RuntimeError('Invalid boolean code {}'.format(hex(value))) + raise RuntimeError(f'Invalid boolean code {hex(value)}') def tgread_date(self): """Reads and converts Unix time (used by Telegram) @@ -131,19 +131,19 @@ def tgread_object(self): return [self.tgread_object() for _ in range(self.read_int())] clazz = core_objects.get(constructor_id, None) - if clazz is None: - # If there was still no luck, give up - self.seek(-4) # Go back - pos = self.tell_position() - error = TypeNotFoundError(constructor_id, self.read()) - self.set_position(pos) - raise error + if clazz is None: + # If there was still no luck, give up + self.seek(-4) # Go back + pos = self.tell_position() + error = TypeNotFoundError(constructor_id, self.read()) + self.set_position(pos) + raise error return clazz.from_reader(self) def tgread_vector(self): """Reads a vector (a list) of Telegram objects.""" - if 0x1cb5c415 != self.read_int(signed=False): + if self.read_int(signed=False) != 0x1CB5C415: raise RuntimeError('Invalid constructor code, vector was expected') count = self.read_int() diff --git a/telethon/extensions/html.py b/telethon/extensions/html.py index 62e622ba0..3c222ce22 100644 --- a/telethon/extensions/html.py +++ b/telethon/extensions/html.py @@ -45,13 +45,13 @@ def handle_starttag(self, tag, attrs): attrs = dict(attrs) EntityType = None args = {} - if tag == 'strong' or tag == 'b': + if tag in ['strong', 'b']: EntityType = MessageEntityBold - elif tag == 'em' or tag == 'i': + elif tag in ['em', 'i']: EntityType = MessageEntityItalic elif tag == 'u': EntityType = MessageEntityUnderline - elif tag == 'del' or tag == 's': + elif tag in ['del', 's']: EntityType = MessageEntityStrike elif tag == 'blockquote': EntityType = MessageEntityBlockquote @@ -81,13 +81,12 @@ def handle_starttag(self, tag, attrs): if url.startswith('mailto:'): url = url[len('mailto:'):] EntityType = MessageEntityEmail + elif self.get_starttag_text() == url: + EntityType = MessageEntityUrl else: - if self.get_starttag_text() == url: - EntityType = MessageEntityUrl - else: - EntityType = MessageEntityTextUrl - args['url'] = url - url = None + EntityType = MessageEntityTextUrl + args['url'] = url + url = None self._open_tags_meta.popleft() self._open_tags_meta.appendleft(url) @@ -101,8 +100,7 @@ def handle_starttag(self, tag, attrs): def handle_data(self, text): previous_tag = self._open_tags[0] if len(self._open_tags) > 0 else '' if previous_tag == 'a': - url = self._open_tags_meta[0] - if url: + if url := self._open_tags_meta[0]: text = url for tag, entity in self._building_entities.items(): @@ -116,8 +114,7 @@ def handle_endtag(self, tag): self._open_tags_meta.popleft() except IndexError: pass - entity = self._building_entities.pop(tag, None) - if entity: + if entity := self._building_entities.pop(tag, None): self.entities.append(entity) @@ -186,17 +183,17 @@ def unparse(text: str, entities: Iterable[TypeMessageEntity], _offset: int = 0, entity_type = type(entity) if entity_type == MessageEntityBold: - html.append('{}'.format(entity_text)) + html.append(f'{entity_text}') elif entity_type == MessageEntityItalic: - html.append('{}'.format(entity_text)) + html.append(f'{entity_text}') elif entity_type == MessageEntityCode: - html.append('{}'.format(entity_text)) + html.append(f'{entity_text}') elif entity_type == MessageEntityUnderline: - html.append('{}'.format(entity_text)) + html.append(f'{entity_text}') elif entity_type == MessageEntityStrike: - html.append('{}'.format(entity_text)) + html.append(f'{entity_text}') elif entity_type == MessageEntityBlockquote: - html.append('
{}
'.format(entity_text)) + html.append(f'
{entity_text}
') elif entity_type == MessageEntityPre: if entity.language: html.append( @@ -206,18 +203,15 @@ def unparse(text: str, entities: Iterable[TypeMessageEntity], _offset: int = 0, " \n" "".format(entity.language, entity_text)) else: - html.append('
{}
' - .format(entity_text)) + html.append(f'
{entity_text}
') elif entity_type == MessageEntityEmail: html.append('{0}'.format(entity_text)) elif entity_type == MessageEntityUrl: html.append('{0}'.format(entity_text)) elif entity_type == MessageEntityTextUrl: - html.append('{}' - .format(escape(entity.url), entity_text)) + html.append(f'{entity_text}') elif entity_type == MessageEntityMentionName: - html.append('{}' - .format(entity.user_id, entity_text)) + html.append(f'{entity_text}') else: skip_entity = True last_offset = relative_offset + (0 if skip_entity else length) diff --git a/telethon/extensions/markdown.py b/telethon/extensions/markdown.py index f6d59106d..22449226f 100644 --- a/telethon/extensions/markdown.py +++ b/telethon/extensions/markdown.py @@ -56,8 +56,13 @@ def parse(message, delimiters=None, url_re=None): # Build a regex to efficiently test all delimiters at once. # Note that the largest delimiter should go first, we don't # want ``` to be interpreted as a single back-tick in a code block. - delim_re = re.compile('|'.join('({})'.format(re.escape(k)) - for k in sorted(delimiters, key=len, reverse=True))) + delim_re = re.compile( + '|'.join( + f'({re.escape(k)})' + for k in sorted(delimiters, key=len, reverse=True) + ) + ) + # Cannot use a for loop because we need to skip some indices i = 0 @@ -67,10 +72,7 @@ def parse(message, delimiters=None, url_re=None): # The offset will just be half the index we're at. message = add_surrogate(message) while i < len(message): - m = delim_re.match(message, pos=i) - - # Did we find some delimiter here at `i`? - if m: + if m := delim_re.match(message, pos=i): delim = next(filter(None, m.groups())) # +1 to avoid matching right after (e.g. "****") @@ -91,11 +93,7 @@ def parse(message, delimiters=None, url_re=None): # If the end is after our start, it is affected if ent.offset + ent.length > i: # If the old start is also before ours, it is fully enclosed - if ent.offset <= i: - ent.length -= len(delim) * 2 - else: - ent.length -= len(delim) - + ent.length -= len(delim) * 2 if ent.offset <= i else len(delim) # Append the found entity ent = delimiters[delim] if ent == MessageEntityPre: @@ -110,8 +108,7 @@ def parse(message, delimiters=None, url_re=None): continue elif url_re: - m = url_re.match(message, pos=i) - if m: + if m := url_re.match(message, pos=i): # Replace the whole match with only the inline URL text. message = ''.join(( message[:m.start()], @@ -167,20 +164,16 @@ def unparse(text, entities, delimiters=None, url_fmt=None): for entity in entities: s = entity.offset e = entity.offset + entity.length - delimiter = delimiters.get(type(entity), None) - if delimiter: - insert_at.append((s, delimiter)) - insert_at.append((e, delimiter)) + if delimiter := delimiters.get(type(entity), None): + insert_at.extend(((s, delimiter), (e, delimiter))) else: url = None if isinstance(entity, MessageEntityTextUrl): url = entity.url elif isinstance(entity, MessageEntityMentionName): - url = 'tg://user?id={}'.format(entity.user_id) + url = f'tg://user?id={entity.user_id}' if url: - insert_at.append((s, '[')) - insert_at.append((e, ']({})'.format(url))) - + insert_at.extend(((s, '['), (e, f']({url})'))) insert_at.sort(key=lambda t: t[0]) while insert_at: at, what = insert_at.pop() diff --git a/telethon/helpers.py b/telethon/helpers.py index 6c782b0b9..9238163a8 100644 --- a/telethon/helpers.py +++ b/telethon/helpers.py @@ -30,8 +30,7 @@ def generate_random_long(signed=True): def ensure_parent_dir_exists(file_path): """Ensures that the parent directory exists""" - parent = os.path.dirname(file_path) - if parent: + if parent := os.path.dirname(file_path): os.makedirs(parent, exist_ok=True) @@ -111,7 +110,7 @@ def retry_range(retries, force_retry=True): # We need at least one iteration even if the retries are 0 # when force_retry is True. - if force_retry and not (retries is None or retries < 0): + if force_retry and retries is not None and retries >= 0: retries += 1 attempt = 0 @@ -122,10 +121,7 @@ def retry_range(retries, force_retry=True): async def _maybe_await(value): - if inspect.isawaitable(value): - return await value - else: - return value + return await value if inspect.isawaitable(value) else value async def _cancel(log, **tasks): @@ -170,11 +166,7 @@ def _sync_enter(self): Helps to cut boilerplate on async context managers that offer synchronous variants. """ - if hasattr(self, 'loop'): - loop = self.loop - else: - loop = self._client.loop - + loop = self.loop if hasattr(self, 'loop') else self._client.loop if loop.is_running(): raise RuntimeError( 'You must use "async with" if the event loop ' @@ -185,11 +177,7 @@ def _sync_enter(self): def _sync_exit(self, *args): - if hasattr(self, 'loop'): - loop = self.loop - else: - loop = self._client.loop - + loop = self.loop if hasattr(self, 'loop') else self._client.loop return loop.run_until_complete(self.__aexit__(*args)) @@ -211,22 +199,24 @@ def _entity_type(entity): 0x1f4661b9, # crc32(b'UserFull') 0xd49a2697, # crc32(b'ChatFull') ): - raise TypeError('{} does not have any entity type'.format(entity)) + raise TypeError(f'{entity} does not have any entity type') except AttributeError: - raise TypeError('{} is not a TLObject, cannot determine entity type'.format(entity)) + raise TypeError(f'{entity} is not a TLObject, cannot determine entity type') name = entity.__class__.__name__ - if 'User' in name: + if ( + 'User' in name + or 'Chat' not in name + and 'Channel' not in name + and 'Self' in name + ): return _EntityType.USER elif 'Chat' in name: return _EntityType.CHAT elif 'Channel' in name: return _EntityType.CHANNEL - elif 'Self' in name: - return _EntityType.USER - # 'Empty' in name or not found, we don't care, not a valid entity. - raise TypeError('{} does not have any entity type'.format(entity)) + raise TypeError(f'{entity} does not have any entity type') # endregion @@ -279,12 +269,10 @@ def __init__(self, *args, **kwargs): self.total = 0 def __str__(self): - return '[{}, total={}]'.format( - ', '.join(str(x) for x in self), self.total) + return f"[{', '.join((str(x) for x in self))}, total={self.total}]" def __repr__(self): - return '[{}, total={}]'.format( - ', '.join(repr(x) for x in self), self.total) + return f"[{', '.join((repr(x) for x in self))}, total={self.total}]" class _FileStream(io.IOBase): diff --git a/telethon/network/authenticator.py b/telethon/network/authenticator.py index ea4762073..b59d7146f 100644 --- a/telethon/network/authenticator.py +++ b/telethon/network/authenticator.py @@ -29,7 +29,7 @@ async def do_authentication(sender): # Step 1 sending: PQ Request, endianness doesn't matter since it's random nonce = int.from_bytes(os.urandom(16), 'big', signed=True) res_pq = await sender.send(ReqPqMultiRequest(nonce)) - assert isinstance(res_pq, ResPQ), 'Step 1 answer was %s' % res_pq + assert isinstance(res_pq, ResPQ), f'Step 1 answer was {res_pq}' if res_pq.nonce != nonce: raise SecurityError('Step 1 invalid nonce from server') @@ -81,8 +81,9 @@ async def do_authentication(sender): )) assert isinstance( - server_dh_params, (ServerDHParamsOk, ServerDHParamsFail)),\ - 'Step 2.1 answer was %s' % server_dh_params + server_dh_params, (ServerDHParamsOk, ServerDHParamsFail) + ), f'Step 2.1 answer was {server_dh_params}' + if server_dh_params.nonce != res_pq.nonce: raise SecurityError('Step 2 invalid nonce from server') @@ -98,8 +99,10 @@ async def do_authentication(sender): if server_dh_params.new_nonce_hash != nnh: raise SecurityError('Step 2 invalid DH fail nonce from server') - assert isinstance(server_dh_params, ServerDHParamsOk),\ - 'Step 2.2 answer was %s' % server_dh_params + assert isinstance( + server_dh_params, ServerDHParamsOk + ), f'Step 2.2 answer was {server_dh_params}' + # Step 3 sending: Complete DH Exchange key, iv = helpers.generate_key_data_from_nonce( @@ -116,8 +119,10 @@ async def do_authentication(sender): with BinaryReader(plain_text_answer) as reader: reader.read(20) # hash sum server_dh_inner = reader.tgread_object() - assert isinstance(server_dh_inner, ServerDHInnerData),\ - 'Step 3 answer was %s' % server_dh_inner + assert isinstance( + server_dh_inner, ServerDHInnerData + ), f'Step 3 answer was {server_dh_inner}' + if server_dh_inner.nonce != res_pq.nonce: raise SecurityError('Step 3 Invalid nonce in encrypted answer') @@ -177,25 +182,24 @@ async def do_authentication(sender): )) nonce_types = (DhGenOk, DhGenRetry, DhGenFail) - assert isinstance(dh_gen, nonce_types), 'Step 3.1 answer was %s' % dh_gen + assert isinstance(dh_gen, nonce_types), f'Step 3.1 answer was {dh_gen}' name = dh_gen.__class__.__name__ if dh_gen.nonce != res_pq.nonce: - raise SecurityError('Step 3 invalid {} nonce from server'.format(name)) + raise SecurityError(f'Step 3 invalid {name} nonce from server') if dh_gen.server_nonce != res_pq.server_nonce: - raise SecurityError( - 'Step 3 invalid {} server nonce from server'.format(name)) + raise SecurityError(f'Step 3 invalid {name} server nonce from server') auth_key = AuthKey(rsa.get_byte_array(gab)) nonce_number = 1 + nonce_types.index(type(dh_gen)) new_nonce_hash = auth_key.calc_new_nonce_hash(new_nonce, nonce_number) - dh_hash = getattr(dh_gen, 'new_nonce_hash{}'.format(nonce_number)) + dh_hash = getattr(dh_gen, f'new_nonce_hash{nonce_number}') if dh_hash != new_nonce_hash: raise SecurityError('Step 3 invalid new nonce hash') if not isinstance(dh_gen, DhGenOk): - raise AssertionError('Step 3.2 answer was %s' % dh_gen) + raise AssertionError(f'Step 3.2 answer was {dh_gen}') return auth_key, time_offset diff --git a/telethon/network/connection/connection.py b/telethon/network/connection/connection.py index 8bff043dc..5a264527c 100644 --- a/telethon/network/connection/connection.py +++ b/telethon/network/connection/connection.py @@ -76,14 +76,14 @@ def _parse_proxy(proxy_type, addr, port, rdns=True, username=None, password=None # We do the check for numerical values here # to be backwards compatible with PySocks proxy format, # (since socks.SOCKS5 == 2, socks.SOCKS4 == 1, socks.HTTP == 3) - if proxy_type == ProxyType.SOCKS5 or proxy_type == 2 or proxy_type == "socks5": + if proxy_type in [ProxyType.SOCKS5, 2, "socks5"]: protocol = ProxyType.SOCKS5 - elif proxy_type == ProxyType.SOCKS4 or proxy_type == 1 or proxy_type == "socks4": + elif proxy_type in [ProxyType.SOCKS4, 1, "socks4"]: protocol = ProxyType.SOCKS4 - elif proxy_type == ProxyType.HTTP or proxy_type == 3 or proxy_type == "http": + elif proxy_type in [ProxyType.HTTP, 3, "http"]: protocol = ProxyType.HTTP else: - raise ValueError("Unknown proxy protocol type: {}".format(proxy_type)) + raise ValueError(f"Unknown proxy protocol type: {proxy_type}") # This tuple must be compatible with `python_socks`' `Proxy.create()` signature return protocol, addr, port, username, password, rdns @@ -91,14 +91,14 @@ def _parse_proxy(proxy_type, addr, port, rdns=True, username=None, password=None else: from socks import SOCKS5, SOCKS4, HTTP - if proxy_type == 2 or proxy_type == "socks5": + if proxy_type in [2, "socks5"]: protocol = SOCKS5 - elif proxy_type == 1 or proxy_type == "socks4": + elif proxy_type in [1, "socks4"]: protocol = SOCKS4 - elif proxy_type == 3 or proxy_type == "http": + elif proxy_type in [3, "http"]: protocol = HTTP else: - raise ValueError("Unknown proxy protocol type: {}".format(proxy_type)) + raise ValueError(f"Unknown proxy protocol type: {proxy_type}") # This tuple must be compatible with `PySocks`' `socksocket.set_proxy()` signature return protocol, addr, port, rdns, username, password @@ -199,19 +199,15 @@ async def _proxy_connect(self, timeout=None, local_addr=None): return sock async def _connect(self, timeout=None, ssl=None): - if self._local_addr is not None: - # NOTE: If port is not specified, we use 0 port - # to notify the OS that port should be chosen randomly - # from the available ones. - if isinstance(self._local_addr, tuple) and len(self._local_addr) == 2: - local_addr = self._local_addr - elif isinstance(self._local_addr, str): - local_addr = (self._local_addr, 0) - else: - raise ValueError("Unknown local address format: {}".format(self._local_addr)) - else: + if self._local_addr is None: local_addr = None + elif isinstance(self._local_addr, tuple) and len(self._local_addr) == 2: + local_addr = self._local_addr + elif isinstance(self._local_addr, str): + local_addr = (self._local_addr, 0) + else: + raise ValueError(f"Unknown local address format: {self._local_addr}") if not self._proxy: self._reader, self._writer = await asyncio.wait_for( asyncio.open_connection( @@ -367,10 +363,7 @@ async def _recv(self): return await self._codec.read_packet(self._reader) def __str__(self): - return '{}:{}/{}'.format( - self._ip, self._port, - self.__class__.__name__.replace('Connection', '') - ) + return f"{self._ip}:{self._port}/{self.__class__.__name__.replace('Connection', '')}" class ObfuscatedConnection(Connection): diff --git a/telethon/network/mtprotosender.py b/telethon/network/mtprotosender.py index ca592ac0a..e6c5662e7 100644 --- a/telethon/network/mtprotosender.py +++ b/telethon/network/mtprotosender.py @@ -225,8 +225,8 @@ async def _connect(self): for attempt in retry_range(self._retries): if not connected: connected = await self._try_connect(attempt) - if not connected: - continue # skip auth key generation until we're connected + if not connected: + continue # skip auth key generation until we're connected if not self.auth_key: try: @@ -250,9 +250,9 @@ async def _connect(self): break # all steps done, break retry loop else: if not connected: - raise ConnectionError('Connection to Telegram failed {} time(s)'.format(self._retries)) + raise ConnectionError(f'Connection to Telegram failed {self._retries} time(s)') - e = ConnectionError('auth_key generation failed {} time(s)'.format(self._retries)) + e = ConnectionError(f'auth_key generation failed {self._retries} time(s)') await self._disconnect(error=e) raise e @@ -473,14 +473,13 @@ async def _send_loop(self): # so even if the network fails they won't be lost. If they were # never re-enqueued, the future waiting for a response "locks". for state in batch: - if not isinstance(state, list): - if isinstance(state.request, TLRequest): - self._pending_state[state.msg_id] = state - else: + if isinstance(state, list): for s in state: if isinstance(s.request, TLRequest): self._pending_state[s.msg_id] = s + elif isinstance(state.request, TLRequest): + self._pending_state[state.msg_id] = state try: await self._connection.send(data) except IOError as e: @@ -560,23 +559,17 @@ def _pop_states(self, msg_id): This method should be used when the response isn't specific. """ - state = self._pending_state.pop(msg_id, None) - if state: + if state := self._pending_state.pop(msg_id, None): return [state] - to_pop = [] - for state in self._pending_state.values(): - if state.container_id == msg_id: - to_pop.append(state.msg_id) - - if to_pop: + if to_pop := [ + state.msg_id + for state in self._pending_state.values() + if state.container_id == msg_id + ]: return [self._pending_state.pop(x) for x in to_pop] - for ack in self._last_acks: - if ack.msg_id == msg_id: - return [ack] - - return [] + return next(([ack] for ack in self._last_acks if ack.msg_id == msg_id), []) async def _handle_rpc_result(self, message): """ @@ -667,8 +660,7 @@ async def _handle_pong(self, message): if self._ping == pong.ping_id: self._ping = None - state = self._pending_state.pop(pong.msg_id, None) - if state: + if state := self._pending_state.pop(pong.msg_id, None): state.future.set_result(pong) async def _handle_bad_server_salt(self, message): @@ -793,8 +785,7 @@ async def _handle_future_salts(self, message): # TODO save these salts and automatically adjust to the # correct one whenever the salt in use expires. self._log.debug('Handling future salts for message %d', message.msg_id) - state = self._pending_state.pop(message.msg_id, None) - if state: + if state := self._pending_state.pop(message.msg_id, None): state.future.set_result(message.obj) async def _handle_state_forgotten(self, message): diff --git a/telethon/network/mtprotostate.py b/telethon/network/mtprotostate.py index 0fe9cc089..84bd5fb4f 100644 --- a/telethon/network/mtprotostate.py +++ b/telethon/network/mtprotostate.py @@ -210,9 +210,8 @@ def _get_seq_no(self, content_related): Generates the next sequence number depending on whether it should be for a content-related query or not. """ - if content_related: - result = self._sequence * 2 + 1 - self._sequence += 1 - return result - else: + if not content_related: return self._sequence * 2 + result = self._sequence * 2 + 1 + self._sequence += 1 + return result diff --git a/telethon/password.py b/telethon/password.py index 0f9502542..f4731bf27 100644 --- a/telethon/password.py +++ b/telethon/password.py @@ -8,8 +8,10 @@ def check_prime_and_good_check(prime: int, g: int): good_prime_bits_count = 2048 if prime < 0 or prime.bit_length() != good_prime_bits_count: - raise ValueError('bad prime count {}, expected {}' - .format(prime.bit_length(), good_prime_bits_count)) + raise ValueError( + f'bad prime count {prime.bit_length()}, expected {good_prime_bits_count}' + ) + # TODO This is awfully slow if factorization.Factorization.factorize(prime)[0] != 1: @@ -17,23 +19,23 @@ def check_prime_and_good_check(prime: int, g: int): if g == 2: if prime % 8 != 7: - raise ValueError('bad g {}, mod8 {}'.format(g, prime % 8)) + raise ValueError(f'bad g {g}, mod8 {prime % 8}') elif g == 3: if prime % 3 != 2: - raise ValueError('bad g {}, mod3 {}'.format(g, prime % 3)) + raise ValueError(f'bad g {g}, mod3 {prime % 3}') elif g == 4: pass elif g == 5: if prime % 5 not in (1, 4): - raise ValueError('bad g {}, mod5 {}'.format(g, prime % 5)) + raise ValueError(f'bad g {g}, mod5 {prime % 5}') elif g == 6: if prime % 24 not in (19, 23): - raise ValueError('bad g {}, mod24 {}'.format(g, prime % 24)) + raise ValueError(f'bad g {g}, mod24 {prime % 24}') elif g == 7: if prime % 7 not in (3, 5, 6): - raise ValueError('bad g {}, mod7 {}'.format(g, prime % 7)) + raise ValueError(f'bad g {g}, mod7 {prime % 7}') else: - raise ValueError('bad g {}'.format(g)) + raise ValueError(f'bad g {g}') prime_sub1_div2 = (prime - 1) // 2 if factorization.Factorization.factorize(prime_sub1_div2)[0] != 1: @@ -61,9 +63,8 @@ def check_prime_and_good(prime_bytes: bytes, g: int): 0x0D, 0x81, 0x15, 0xF6, 0x35, 0xB1, 0x05, 0xEE, 0x2E, 0x4E, 0x15, 0xD0, 0x4B, 0x24, 0x54, 0xBF, 0x6F, 0x4F, 0xAD, 0xF0, 0x34, 0xB1, 0x04, 0x03, 0x11, 0x9C, 0xD8, 0xE3, 0xB9, 0x2F, 0xCC, 0x5B)) - if good_prime == prime_bytes: - if g in (3, 4, 5, 7): - return # It's good + if good_prime == prime_bytes and g in {3, 4, 5, 7}: + return # It's good check_prime_and_good_check(int.from_bytes(prime_bytes, 'big'), g) diff --git a/telethon/sessions/memory.py b/telethon/sessions/memory.py index 1b1a6bfb0..b44b5a7e0 100644 --- a/telethon/sessions/memory.py +++ b/telethon/sessions/memory.py @@ -138,10 +138,7 @@ def _entities_to_rows(self, tlo): entities.extend(tlo.users) rows = [] # Rows to add (id, hash, username, phone, name) - for e in entities: - row = self._entity_to_row(e) - if row: - rows.append(row) + rows.extend(row for e in entities if (row := self._entity_to_row(e))) return rows def process_entities(self, tlo): @@ -173,14 +170,13 @@ def get_entity_rows_by_id(self, id, exact=True): if exact: return next((id, hash) for found_id, hash, _, _, _ in self._entities if found_id == id) - else: - ids = ( - utils.get_peer_id(PeerUser(id)), - utils.get_peer_id(PeerChat(id)), - utils.get_peer_id(PeerChannel(id)) - ) - return next((id, hash) for found_id, hash, _, _, _ - in self._entities if found_id in ids) + ids = ( + utils.get_peer_id(PeerUser(id)), + utils.get_peer_id(PeerChat(id)), + utils.get_peer_id(PeerChannel(id)) + ) + return next((id, hash) for found_id, hash, _, _, _ + in self._entities if found_id in ids) except StopIteration: pass @@ -202,17 +198,14 @@ def get_input_entity(self, key): result = None if isinstance(key, str): - phone = utils.parse_phone(key) - if phone: + if phone := utils.parse_phone(key): result = self.get_entity_rows_by_phone(phone) else: username, invite = utils.parse_username(key) if username and not invite: result = self.get_entity_rows_by_username(username) - else: - tup = utils.resolve_invite_link(key)[1] - if tup: - result = self.get_entity_rows_by_id(tup, exact=False) + elif tup := utils.resolve_invite_link(key)[1]: + result = self.get_entity_rows_by_id(tup, exact=False) elif isinstance(key, int): result = self.get_entity_rows_by_id(key, exact) @@ -220,22 +213,21 @@ def get_input_entity(self, key): if not result and isinstance(key, str): result = self.get_entity_rows_by_name(key) - if result: - entity_id, entity_hash = result # unpack resulting tuple - entity_id, kind = utils.resolve_id(entity_id) - # removes the mark and returns type of entity - if kind == PeerUser: - return InputPeerUser(entity_id, entity_hash) - elif kind == PeerChat: - return InputPeerChat(entity_id) - elif kind == PeerChannel: - return InputPeerChannel(entity_id, entity_hash) - else: + if not result: raise ValueError('Could not find input entity with key ', key) + entity_id, entity_hash = result # unpack resulting tuple + entity_id, kind = utils.resolve_id(entity_id) + # removes the mark and returns type of entity + if kind == PeerUser: + return InputPeerUser(entity_id, entity_hash) + elif kind == PeerChat: + return InputPeerChat(entity_id) + elif kind == PeerChannel: + return InputPeerChannel(entity_id, entity_hash) def cache_file(self, md5_digest, file_size, instance): if not isinstance(instance, (InputDocument, InputPhoto)): - raise TypeError('Cannot cache %s instance' % type(instance)) + raise TypeError(f'Cannot cache {type(instance)} instance') key = (md5_digest, file_size, _SentFileType.from_type(type(instance))) value = (instance.id, instance.access_hash) self._files[key] = value diff --git a/telethon/sessions/sqlite.py b/telethon/sessions/sqlite.py index 82b2129da..902c7f106 100644 --- a/telethon/sessions/sqlite.py +++ b/telethon/sessions/sqlite.py @@ -59,8 +59,7 @@ def __init__(self, session_id=None): # These values will be saved c.execute('select * from sessions') - tuple_ = c.fetchone() - if tuple_: + if tuple_ := c.fetchone(): self._dc_id, self._server_address, self._port, key, \ self._takeout_id = tuple_ self._auth_key = AuthKey(data=key) @@ -159,7 +158,7 @@ def _upgrade_database(self, old): @staticmethod def _create_table(c, *definitions): for definition in definitions: - c.execute('create table {}'.format(definition)) + c.execute(f'create table {definition}') # Data from sessions should be kept as properties # not to fetch the database every time we need it @@ -169,10 +168,7 @@ def set_dc(self, dc_id, server_address, port): # Fetch the auth_key corresponding to this data center row = self._execute('select auth_key from sessions') - if row and row[0]: - self._auth_key = AuthKey(data=row[0]) - else: - self._auth_key = None + self._auth_key = AuthKey(data=row[0]) if row and row[0] else None @MemorySession.auth_key.setter def auth_key(self, value): @@ -202,9 +198,10 @@ def _update_session_table(self): c.close() def get_update_state(self, entity_id): - row = self._execute('select pts, qts, date, seq from update_state ' - 'where id = ?', entity_id) - if row: + if row := self._execute( + 'select pts, qts, date, seq from update_state ' 'where id = ?', + entity_id, + ): pts, qts, date, seq = row date = datetime.datetime.fromtimestamp( date, tz=datetime.timezone.utc) @@ -242,11 +239,10 @@ def _execute(self, stmt, *values): def close(self): """Closes the connection unless we're working in-memory""" - if self.filename != ':memory:': - if self._conn is not None: - self._conn.commit() - self._conn.close() - self._conn = None + if self.filename != ':memory:' and self._conn is not None: + self._conn.commit() + self._conn.close() + self._conn = None def delete(self): """Deletes the current session file""" @@ -333,18 +329,19 @@ def get_entity_rows_by_id(self, id, exact=True): # File processing def get_file(self, md5_digest, file_size, cls): - row = self._execute( + if row := self._execute( 'select id, hash from sent_files ' 'where md5_digest = ? and file_size = ? and type = ?', - md5_digest, file_size, _SentFileType.from_type(cls).value - ) - if row: + md5_digest, + file_size, + _SentFileType.from_type(cls).value, + ): # Both allowed classes have (id, access_hash) as parameters return cls(row[0], row[1]) def cache_file(self, md5_digest, file_size, instance): if not isinstance(instance, (InputDocument, InputPhoto)): - raise TypeError('Cannot cache %s instance' % type(instance)) + raise TypeError(f'Cannot cache {type(instance)} instance') self._execute( 'insert or replace into sent_files values (?,?,?,?,?)', diff --git a/telethon/statecache.py b/telethon/statecache.py index 0e02bbd49..888f73096 100644 --- a/telethon/statecache.py +++ b/telethon/statecache.py @@ -35,10 +35,7 @@ def __init__(self, initial, loggers): # is lightweight and immutable we can easily copy them around to # each update in case they need to fetch missing entities. self._logger = loggers[__name__] - if initial: - self._pts_date = initial.pts, initial.date - else: - self._pts_date = None, None + self._pts_date = (initial.pts, initial.date) if initial else (None, None) def reset(self): self.__dict__.clear() @@ -152,10 +149,7 @@ def __getitem__(self, item): If no information is known, ``pts`` will be `None`. """ - if item is None: - return self._pts_date - else: - return self.__dict__.get(item) + return self._pts_date if item is None else self.__dict__.get(item) def __setitem__(self, where, value): if where is None: diff --git a/telethon/sync.py b/telethon/sync.py index 80b80beac..081351e24 100644 --- a/telethon/sync.py +++ b/telethon/sync.py @@ -33,10 +33,7 @@ def _syncify_wrap(t, method_name): def syncified(*args, **kwargs): coro = method(*args, **kwargs) loop = asyncio.get_event_loop() - if loop.is_running(): - return coro - else: - return loop.run_until_complete(coro) + return coro if loop.is_running() else loop.run_until_complete(coro) # Save an accessible reference to the original method setattr(syncified, '__tl.sync', method) @@ -54,9 +51,10 @@ def syncify(*types): # about asyncgenfunction's here. for t in types: for name in dir(t): - if not name.startswith('_') or name == '__call__': - if inspect.iscoroutinefunction(getattr(t, name)): - _syncify_wrap(t, name) + if ( + not name.startswith('_') or name == '__call__' + ) and inspect.iscoroutinefunction(getattr(t, name)): + _syncify_wrap(t, name) syncify(TelegramClient, _TakeoutClient, Draft, Dialog, MessageButton, diff --git a/telethon/tl/core/gzippacked.py b/telethon/tl/core/gzippacked.py index fb4094e41..364c57cb9 100644 --- a/telethon/tl/core/gzippacked.py +++ b/telethon/tl/core/gzippacked.py @@ -18,11 +18,10 @@ def gzip_if_smaller(content_related, data): Note that this only applies to content related requests. """ - if content_related and len(data) > 512: - gzipped = bytes(GzipPacked(data)) - return gzipped if len(gzipped) < len(data) else data - else: + if not content_related or len(data) <= 512: return data + gzipped = bytes(GzipPacked(data)) + return gzipped if len(gzipped) < len(data) else data def __bytes__(self): return struct.pack('`. """ if message is None: - if self._incoming: - message = self._incoming[-1].id - else: - message = 0 + message = self._incoming[-1].id if self._incoming else 0 elif not isinstance(message, int): message = message.id @@ -199,12 +192,14 @@ def _get_message( # If there is no last-chosen ID, make sure to pick one *after* # the input message, since we don't want responses back in time if target_id not in indices: - for i, incoming in enumerate(self._incoming): - if incoming.id > target_id: - indices[target_id] = i - break - else: - indices[target_id] = len(self._incoming) + indices[target_id] = next( + ( + i + for i, incoming in enumerate(self._incoming) + if incoming.id > target_id + ), + len(self._incoming), + ) # We will always return a future from here, even if the result # can be set immediately. Otherwise, needing to await only @@ -346,9 +341,7 @@ async def main(): async def _check_custom(self, built): for key, (ev, fut) in list(self._custom.items()): ev_type = type(ev) - inst = built[ev_type] - - if inst: + if inst := built[ev_type]: filter = ev.filter(inst) if inspect.isawaitable(filter): filter = await filter diff --git a/telethon/tl/custom/inlinebuilder.py b/telethon/tl/custom/inlinebuilder.py index f82b7e677..d6dd93279 100644 --- a/telethon/tl/custom/inlinebuilder.py +++ b/telethon/tl/custom/inlinebuilder.py @@ -299,8 +299,8 @@ async def document( type = ty break - if type is None: - type = 'file' + if type is None: + type = 'file' try: fh = utils.get_input_document(file) diff --git a/telethon/tl/custom/inlineresults.py b/telethon/tl/custom/inlineresults.py index 9ab0c0a83..f604b840f 100644 --- a/telethon/tl/custom/inlineresults.py +++ b/telethon/tl/custom/inlineresults.py @@ -65,16 +65,7 @@ def results_valid(self): return time.time() < self._valid_until def _to_str(self, item_function): - return ('[{}, query_id={}, cache_time={}, users={}, gallery={}, ' - 'next_offset={}, switch_pm={}]'.format( - ', '.join(item_function(x) for x in self), - self.query_id, - self.cache_time, - self.users, - self.gallery, - self.next_offset, - self.switch_pm - )) + return f"[{', '.join((item_function(x) for x in self))}, query_id={self.query_id}, cache_time={self.cache_time}, users={self.users}, gallery={self.gallery}, next_offset={self.next_offset}, switch_pm={self.switch_pm}]" def __str__(self): return self._to_str(str) diff --git a/telethon/tl/custom/message.py b/telethon/tl/custom/message.py index b55f90652..a2721060f 100644 --- a/telethon/tl/custom/message.py +++ b/telethon/tl/custom/message.py @@ -337,11 +337,11 @@ def text(self): parse mode. Will be `None` for :tl:`MessageService`. """ if self._text is None and self._client: - if not self._client.parse_mode: - self._text = self.message - else: - self._text = self._client.parse_mode.unparse( - self.message, self.entities) + self._text = ( + self._client.parse_mode.unparse(self.message, self.entities) + if self._client.parse_mode + else self.message + ) return self._text @@ -456,8 +456,7 @@ def file(self): etc., without having to manually inspect the ``document.attributes``. """ if not self._file: - media = self.photo or self.document - if media: + if media := self.photo or self.document: self._file = File(media) return self._file @@ -499,9 +498,10 @@ def web_preview(self): """ The :tl:`WebPage` media in this message, if any. """ - if isinstance(self.media, types.MessageMediaWebPage): - if isinstance(self.media.webpage, types.WebPage): - return self.media.webpage + if isinstance(self.media, types.MessageMediaWebPage) and isinstance( + self.media.webpage, types.WebPage + ): + return self.media.webpage @property def audio(self): @@ -966,14 +966,14 @@ def find_options(): return [answers[idx].option for idx in i] return [answers[i].option] if text is not None: - if callable(text): - for answer in answers: - if text(answer.text): - return [answer.option] - else: - for answer in answers: - if answer.text == text: - return [answer.option] + for answer in answers: + if ( + callable(text) + and text(answer.text) + or not callable(text) + and answer.text == text + ): + return [answer.option] return if filter is not None: @@ -999,14 +999,14 @@ def find_options(): def find_button(): nonlocal i if text is not None: - if callable(text): - for button in self._buttons_flat: - if text(button.text): - return button - else: - for button in self._buttons_flat: - if button.text == text: - return button + for button in self._buttons_flat: + if ( + callable(text) + and text(button.text) + or not callable(text) + and button.text == text + ): + return button return if filter is not None: @@ -1017,10 +1017,7 @@ def find_button(): if i is None: i = 0 - if j is None: - return self._buttons_flat[i] - else: - return self._buttons[i][j] + return self._buttons_flat[i] if j is None else self._buttons[i][j] button = find_button() if button: @@ -1137,12 +1134,9 @@ def _document_by_attribute(self, kind, condition=None): Helper method to return the document only if it has an attribute that's an instance of the given kind, and passes the condition. """ - doc = self.document - if doc: + if doc := self.document: for attr in doc.attributes: if isinstance(attr, kind): - if not condition or condition(attr): - return doc - return None + return doc if not condition or condition(attr) else None # endregion Private Methods diff --git a/telethon/tl/custom/qrlogin.py b/telethon/tl/custom/qrlogin.py index 39585f2f9..2780356f3 100644 --- a/telethon/tl/custom/qrlogin.py +++ b/telethon/tl/custom/qrlogin.py @@ -54,7 +54,7 @@ def url(self) -> str: The URL simply consists of `token` base64-encoded. """ - return 'tg://login?token={}'.format(base64.urlsafe_b64encode(self._resp.token).decode('utf-8').rstrip('=')) + return f"tg://login?token={base64.urlsafe_b64encode(self._resp.token).decode('utf-8').rstrip('=')}" @property def expires(self) -> datetime.datetime: diff --git a/telethon/tl/tlobject.py b/telethon/tl/tlobject.py index 4b94e00f0..b80788078 100644 --- a/telethon/tl/tlobject.py +++ b/telethon/tl/tlobject.py @@ -41,65 +41,59 @@ def pretty_format(obj, indent=None): Pretty formats the given object as a string which is returned. If indent is None, a single line will be returned. """ - if indent is None: - if isinstance(obj, TLObject): - obj = obj.to_dict() + if isinstance(obj, TLObject): + obj = obj.to_dict() + if indent is None: if isinstance(obj, dict): - return '{}({})'.format(obj.get('_', 'dict'), ', '.join( - '{}={}'.format(k, TLObject.pretty_format(v)) - for k, v in obj.items() if k != '_' - )) - elif isinstance(obj, str) or isinstance(obj, bytes): - return repr(obj) - elif hasattr(obj, '__iter__'): - return '[{}]'.format( - ', '.join(TLObject.pretty_format(x) for x in obj) + return '{}({})'.format( + obj.get('_', 'dict'), + ', '.join( + f'{k}={TLObject.pretty_format(v)}' + for k, v in obj.items() + if k != '_' + ), ) - else: + + elif isinstance(obj, (str, bytes)) or not hasattr(obj, '__iter__'): return repr(obj) + else: + return f"[{', '.join((TLObject.pretty_format(x) for x in obj))}]" else: result = [] - if isinstance(obj, TLObject): - obj = obj.to_dict() - if isinstance(obj, dict): - result.append(obj.get('_', 'dict')) - result.append('(') + result.extend((obj.get('_', 'dict'), '(')) if obj: result.append('\n') indent += 1 for k, v in obj.items(): if k == '_': continue - result.append('\t' * indent) - result.append(k) - result.append('=') - result.append(TLObject.pretty_format(v, indent)) - result.append(',\n') + result.extend( + ( + '\t' * indent, + k, + '=', + TLObject.pretty_format(v, indent), + ',\n', + ) + ) + result.pop() # last ',\n' indent -= 1 - result.append('\n') - result.append('\t' * indent) + result.extend(('\n', '\t' * indent)) result.append(')') - elif isinstance(obj, str) or isinstance(obj, bytes): + elif isinstance(obj, (str, bytes)) or not hasattr(obj, '__iter__'): result.append(repr(obj)) - elif hasattr(obj, '__iter__'): + else: result.append('[\n') indent += 1 for x in obj: - result.append('\t' * indent) - result.append(TLObject.pretty_format(x, indent)) - result.append(',\n') + result.extend(('\t' * indent, TLObject.pretty_format(x, indent), ',\n')) indent -= 1 - result.append('\t' * indent) - result.append(']') - - else: - result.append(repr(obj)) - + result.extend(('\t' * indent, ']')) return ''.join(result) @staticmethod @@ -109,8 +103,7 @@ def serialize_bytes(data): if isinstance(data, str): data = data.encode('utf-8') else: - raise TypeError( - 'bytes or str expected, not {}'.format(type(data))) + raise TypeError(f'bytes or str expected, not {type(data)}') r = [] if len(data) < 254: @@ -119,8 +112,6 @@ def serialize_bytes(data): padding = 4 - padding r.append(bytes([len(data)])) - r.append(data) - else: padding = len(data) % 4 if padding != 0: @@ -132,9 +123,7 @@ def serialize_bytes(data): (len(data) >> 8) % 256, (len(data) >> 16) % 256 ])) - r.append(data) - - r.append(bytes(padding)) + r.extend((data, bytes(padding))) return b''.join(r) @staticmethod @@ -155,7 +144,7 @@ def serialize_datetime(dt): if isinstance(dt, int): return struct.pack(' 1000000000000: - marked_id -= 1000000000000 - return marked_id, types.PeerChannel - else: + if marked_id <= 1000000000000: return marked_id, types.PeerChat + marked_id -= 1000000000000 + return marked_id, types.PeerChannel def _rle_decode(data): @@ -1158,19 +1143,18 @@ def resolve_bot_file_id(file_id): return None attributes = [] - if file_type == 3 or file_type == 9: + if file_type in [3, 9]: attributes.append(types.DocumentAttributeAudio( duration=0, voice=file_type == 3 )) - elif file_type == 4 or file_type == 13: + elif file_type in [4, 13]: attributes.append(types.DocumentAttributeVideo( duration=0, w=0, h=0, round_message=file_type == 13 )) - # elif file_type == 5: # other, cannot know which elif file_type == 8: attributes.append(types.DocumentAttributeSticker( alt='', @@ -1190,7 +1174,11 @@ def resolve_bot_file_id(file_id): attributes=attributes, file_reference=b'' ) - elif (version == 2 and len(data) == 44) or (version == 4 and len(data) in (49, 77)): + elif ( + (version == 2 and len(data) == 44) + or version == 4 + and len(data) in {49, 77} + ): if version == 2: (file_type, dc_id, media_id, access_hash, volume_id, secret, local_id) = struct.unpack('LQ', payload)) elif len(payload) == 16: return struct.unpack('>LLQ', payload) - else: - pass except (struct.error, TypeError): pass return None, None, None diff --git a/telethon_examples/gui.py b/telethon_examples/gui.py index bd241f606..4b741eace 100644 --- a/telethon_examples/gui.py +++ b/telethon_examples/gui.py @@ -168,7 +168,7 @@ async def on_message(self, event): # If the message has media show "(MediaType) " if event.media: - text += '({}) '.format(event.media.__class__.__name__) + text += f'({event.media.__class__.__name__}) ' text += sanitize_str(event.text) text += '\n' @@ -247,12 +247,7 @@ async def send_message(self, event=None): if not text: return - # NOTE: This part is optional but supports editing messages - # You can remove it if you find it too complicated. - # - # Check if the edit matches any text - m = EDIT.match(text) - if m: + if m := EDIT.match(text): find = re.compile(m.group(1).lstrip()) # Cannot reversed(enumerate(...)), use index for i in reversed(range(len(self.sent_text))): @@ -269,9 +264,7 @@ async def send_message(self, event=None): self.log.yview(tkinter.END) return - # Check if we want to delete the message - m = DELETE.match(text) - if m: + if m := DELETE.match(text): try: delete = self.message_ids.pop(-int(m.group(1))) except IndexError: @@ -285,8 +278,7 @@ async def send_message(self, event=None): # Check if we want to reply to some message reply_to = None - m = REPLY.match(text) - if m: + if m := REPLY.match(text): text = m.group(2) try: reply_to = self.message_ids[-int(m.group(1))] diff --git a/telethon_examples/interactive_telegram_client.py b/telethon_examples/interactive_telegram_client.py index 88f491de6..2c604b9ce 100644 --- a/telethon_examples/interactive_telegram_client.py +++ b/telethon_examples/interactive_telegram_client.py @@ -26,9 +26,9 @@ def sprint(string, *args, **kwargs): def print_title(title): """Helper function to print titles to the console more nicely""" sprint('\n') - sprint('=={}=='.format('=' * len(title))) - sprint('= {} ='.format(title)) - sprint('=={}=='.format('=' * len(title))) + sprint(f"=={'=' * len(title)}==") + sprint(f'= {title} =') + sprint(f"=={'=' * len(title)}==") def bytes_to_string(byte_count): @@ -158,13 +158,13 @@ async def run(self): # are much easier to use. self.add_event_handler(self.message_handler, events.NewMessage) + # Retrieve the top dialogs. You can set the limit to None to + # retrieve all of them if you wish, but beware that may take + # a long time if you have hundreds of them. + dialog_count = 15 + # Enter a while loop to chat as long as the user wants while True: - # Retrieve the top dialogs. You can set the limit to None to - # retrieve all of them if you wish, but beware that may take - # a long time if you have hundreds of them. - dialog_count = 15 - # Entities represent the user, chat or channel # corresponding to the dialog on the same index. dialogs = await self.get_dialogs(limit=dialog_count) @@ -175,7 +175,7 @@ async def run(self): # Display them so the user can choose for i, dialog in enumerate(dialogs, start=1): - sprint('{}. {}'.format(i, get_display_name(dialog.entity))) + sprint(f'{i}. {get_display_name(dialog.entity)}') # Let the user decide who they want to talk to print() @@ -198,7 +198,7 @@ async def run(self): return try: - i = int(i if i else 0) - 1 + i = int(i or 0) - 1 # Ensure it is inside the bounds, otherwise retry if not 0 <= i < dialog_count: i = None @@ -209,7 +209,7 @@ async def run(self): entity = dialogs[i].entity # Show some information - print_title('Chat with "{}"'.format(get_display_name(entity))) + print_title(f'Chat with "{get_display_name(entity)}"') print('Available commands:') print(' !q: Quits the current chat.') print(' !Q: Quits the current chat and exits.') @@ -231,7 +231,6 @@ async def run(self): elif msg == '!Q': return - # History elif msg == '!h': # First retrieve the messages and some information messages = await self.get_messages(entity, limit=10) @@ -249,8 +248,7 @@ async def run(self): # Format the message content if getattr(msg, 'media', None): self.found_media[msg.id] = msg - content = '<{}> {}'.format( - type(msg.media).__name__, msg.message) + content = f'<{type(msg.media).__name__}> {msg.message}' elif hasattr(msg, 'message'): content = msg.message @@ -261,34 +259,28 @@ async def run(self): content = type(msg).__name__ # And print it to the user - sprint('[{}:{}] (ID={}) {}: {}'.format( - msg.date.hour, msg.date.minute, msg.id, name, content)) + sprint(f'[{msg.date.hour}:{msg.date.minute}] (ID={msg.id}) {name}: {content}') - # Send photo elif msg.startswith('!up '): # Slice the message to get the path path = msg[len('!up '):] await self.send_photo(path=path, entity=entity) - # Send file (document) elif msg.startswith('!uf '): # Slice the message to get the path path = msg[len('!uf '):] await self.send_document(path=path, entity=entity) - # Delete messages elif msg.startswith('!d '): # Slice the message to get message ID msg = msg[len('!d '):] deleted_msg = await self.delete_messages(entity, msg) - print('Deleted {}'.format(deleted_msg)) + print(f'Deleted {deleted_msg}') - # Download media elif msg.startswith('!dm '): # Slice the message to get message ID await self.download_media_by_id(msg[len('!dm '):]) - # Download profile photo elif msg == '!dp': print('Downloading profile picture to usermedia/...') os.makedirs('usermedia', exist_ok=True) @@ -305,7 +297,6 @@ async def run(self): for name, val in attributes: print("{:<{width}} : {}".format(name, val, width=pad)) - # Send chat message (if any) elif msg: await self.send_message(entity, msg, link_preview=False) @@ -344,7 +335,7 @@ async def download_media_by_id(self, media_id): file='usermedia/', progress_callback=self.download_progress_callback ) - print('Media downloaded to {}!'.format(output)) + print(f'Media downloaded to {output}!') @staticmethod def download_progress_callback(downloaded_bytes, total_bytes): @@ -377,24 +368,16 @@ async def message_handler(self, event): chat = await event.get_chat() if event.is_group: if event.out: - sprint('>> sent "{}" to chat {}'.format( - event.text, get_display_name(chat) - )) + sprint(f'>> sent "{event.text}" to chat {get_display_name(chat)}') else: - sprint('<< {} @ {} sent "{}"'.format( - get_display_name(await event.get_sender()), - get_display_name(chat), - event.text - )) + sprint( + f'<< {get_display_name(await event.get_sender())} @ {get_display_name(chat)} sent "{event.text}"' + ) + + elif event.out: + sprint(f'>> "{event.text}" to user {get_display_name(chat)}') else: - if event.out: - sprint('>> "{}" to user {}'.format( - event.text, get_display_name(chat) - )) - else: - sprint('<< {} sent "{}"'.format( - get_display_name(chat), event.text - )) + sprint(f'<< {get_display_name(chat)} sent "{event.text}"') if __name__ == '__main__': diff --git a/telethon_examples/payment.py b/telethon_examples/payment.py index 22358a6a4..bed59e2d8 100644 --- a/telethon_examples/payment.py +++ b/telethon_examples/payment.py @@ -47,7 +47,11 @@ def get_env(name, message, cast=str): # If we don't `SetBotPrecheckoutResultsRequest`, money won't be charged from buyer, and nothing will happen next. @bot.on(events.Raw(types.UpdateBotPrecheckoutQuery)) async def payment_pre_checkout_handler(event: types.UpdateBotPrecheckoutQuery): - if event.payload.decode('UTF-8') == 'product A': + if ( + event.payload.decode('UTF-8') == 'product A' + or event.payload.decode('UTF-8') != 'product A' + and event.payload.decode('UTF-8') == 'product B' + ): # so we have to confirm payment await bot( functions.messages.SetBotPrecheckoutResultsRequest( @@ -56,15 +60,6 @@ async def payment_pre_checkout_handler(event: types.UpdateBotPrecheckoutQuery): error=None ) ) - elif event.payload.decode('UTF-8') == 'product B': - # same for another - await bot( - functions.messages.SetBotPrecheckoutResultsRequest( - query_id=event.query_id, - success=True, - error=None - ) - ) else: # for example, something went wrong (whatever reason). We can tell customer about that: await bot( diff --git a/telethon_examples/quart_login.py b/telethon_examples/quart_login.py index 98fb35de6..f3cdf8568 100644 --- a/telethon_examples/quart_login.py +++ b/telethon_examples/quart_login.py @@ -9,9 +9,7 @@ def get_env(name, message): - if name in os.environ: - return os.environ[name] - return input(message) + return os.environ[name] if name in os.environ else input(message) BASE_TEMPLATE = ''' @@ -64,19 +62,13 @@ def get_env(name, message): # Helper method to format messages nicely async def format_message(message): if message.photo: - content = '{}'.format( - base64.b64encode(await message.download_media(bytes)).decode(), - message.raw_text - ) + content = f'{message.raw_text}' + else: # client.parse_mode = 'html', so bold etc. will work! content = (message.text or '(action message)').replace('\n', '
') - return '

{}: {}{}

'.format( - utils.get_display_name(message.sender), - content, - message.date - ) + return f'

{utils.get_display_name(message.sender)}: {content}{message.date}

' # Connect the client before we start serving with Quart @@ -115,7 +107,7 @@ async def root(): if await client.is_user_authorized(): # They are logged in, show them some messages from their first dialog dialog = (await client.get_dialogs())[0] - result = '

{}

'.format(dialog.title) + result = f'

{dialog.title}

' async for m in client.iter_messages(dialog, 10): result += await(format_message(m)) diff --git a/telethon_examples/replier.py b/telethon_examples/replier.py index e498bad9b..d48c7f289 100755 --- a/telethon_examples/replier.py +++ b/telethon_examples/replier.py @@ -67,22 +67,21 @@ async def handler(event): await event.reply('> chrome\nneeds more firefox') # Reply always responds as a reply. We can respond without replying too - if 'shrug' in event.raw_text: - if can_react(event.chat_id): - await event.respond(r'¯\_(ツ)_/¯') - - # We can also use client methods from here - client = event.client + if 'shrug' in event.raw_text and can_react(event.chat_id): + await event.respond(r'¯\_(ツ)_/¯') # If we sent the message, we are replying to someone, # and we said "save pic" in the message if event.out and event.is_reply and 'save pic' in event.raw_text: + # We can also use client methods from here + client = event.client + reply_msg = await event.get_reply_message() replied_to_user = await reply_msg.get_input_sender() message = await event.reply('Downloading your profile photo...') file = await client.download_profile_photo(replied_to_user) - await message.edit('I saved your photo in {}'.format(file)) + await message.edit(f'I saved your photo in {file}') client = TelegramClient( diff --git a/telethon_generator/docswriter.py b/telethon_generator/docswriter.py index 8f626c98c..985c6cf34 100644 --- a/telethon_generator/docswriter.py +++ b/telethon_generator/docswriter.py @@ -71,8 +71,7 @@ def set_menu_separator(self, img): Must be called before adding entries to the menu """ if img: - self.menu_separator_tag = '/'.format( - self._rel(img)) + self.menu_separator_tag = f'/' else: self.menu_separator_tag = None @@ -177,21 +176,20 @@ def write_code(self, tlobject): if tlobject.result == generic_name: # Generic results cannot have any link self.write(tlobject.result) + elif re.search('^vector<', tlobject.result, re.IGNORECASE): + # Notice that we don't simply make up the "Vector" part, + # because some requests (as of now, only FutureSalts), + # use a lower type name for it (see #81) + vector, inner = tlobject.result.split('<') + inner = inner.strip('>') + self.write('{}<', + self.type_to_path(vector), vector) + + self.write('{}>', + self.type_to_path(inner), inner) else: - if re.search('^vector<', tlobject.result, re.IGNORECASE): - # Notice that we don't simply make up the "Vector" part, - # because some requests (as of now, only FutureSalts), - # use a lower type name for it (see #81) - vector, inner = tlobject.result.split('<') - inner = inner.strip('>') - self.write('{}<', - self.type_to_path(vector), vector) - - self.write('{}>', - self.type_to_path(inner), inner) - else: - self.write('{}', - self.type_to_path(tlobject.result), tlobject.result) + self.write('{}', + self.type_to_path(tlobject.result), tlobject.result) self.write('') @@ -249,15 +247,13 @@ def write_copy_button(self, text, text_to_copy): """Writes a button with 'text' which can be used to copy 'text_to_copy' to clipboard when it's clicked.""" self.write_copy_script = True - self.write('' - .format(text_to_copy, text)) + self.write(f"""""") def add_script(self, src='', path=None): if path: - self._script += ''.format( - self._rel(path)) + self._script += f'' elif src: - self._script += ''.format(src) + self._script += f'' def end_body(self): """Ends the whole document. This should be called the last""" diff --git a/telethon_generator/generators/docs.py b/telethon_generator/generators/docs.py index 34b599ffe..273b054d5 100755 --- a/telethon_generator/generators/docs.py +++ b/telethon_generator/generators/docs.py @@ -23,15 +23,14 @@ def _get_file_name(tlobject): # Courtesy of http://stackoverflow.com/a/1176023/4759433 s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) result = re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower() - return '{}.html'.format(result) + return f'{result}.html' def get_import_code(tlobject): """``TLObject -> from ... import ...``.""" kind = 'functions' if tlobject.is_function else 'types' - ns = '.' + tlobject.namespace if tlobject.namespace else '' - return 'from telethon.tl.{}{} import {}'\ - .format(kind, ns, tlobject.class_name) + ns = f'.{tlobject.namespace}' if tlobject.namespace else '' + return f'from telethon.tl.{kind}{ns} import {tlobject.class_name}' def _get_path_for(tlobject): @@ -46,7 +45,7 @@ def _get_path_for(tlobject): def _get_path_for_type(type_): """Similar to `_get_path_for` but for only type names.""" if type_.lower() in CORE_TYPES: - return Path('index.html#%s' % type_.lower()) + return Path(f'index.html#{type_.lower()}') elif '.' in type_: namespace, name = type_.split('.') return Path('types', namespace, _get_file_name(name)) @@ -117,12 +116,15 @@ def _generate_index(folder, paths, .replace(os.path.sep, '/').title()) if bots_index: - docs.write_text('These are the methods that you may be able to ' - 'use as a bot. Click here to ' - 'view them all.'.format(INDEX)) + docs.write_text( + f'These are the methods that you may be able to use as a bot. Click here to view them all.' + ) + else: - docs.write_text('Click here to view the methods ' - 'that you can use as a bot.'.format(BOT_INDEX)) + docs.write_text( + f'Click here to view the methods that you can use as a bot.' + ) + if namespaces: docs.write_title('Namespaces', level=3) docs.begin_table(4) @@ -131,9 +133,11 @@ def _generate_index(folder, paths, # For every namespace, also write the index of it namespace_paths = [] if bots_index: - for item in bots_index_paths: - if item.parent == namespace: - namespace_paths.append(item) + namespace_paths.extend( + item + for item in bots_index_paths + if item.parent == namespace + ) _generate_index(namespace, paths, bots_index, namespace_paths) @@ -576,17 +580,17 @@ def fmt(xs): for x in xs: zs[x.class_name] = x.class_name in zs return ', '.join( - '"{}.{}"'.format(x.namespace, x.class_name) + f'"{x.namespace}.{x.class_name}"' if zs[x.class_name] and x.namespace - else '"{}"'.format(x.class_name) for x in xs + else f'"{x.class_name}"' + for x in xs ) request_names = fmt(methods) constructor_names = fmt(cs) def fmt(xs, formatter): - return ', '.join('"{}"'.format( - formatter(x)).replace(os.path.sep, '/') for x in xs) + return ', '.join(f'"{formatter(x)}"'.replace(os.path.sep, '/') for x in xs) type_names = fmt(types, formatter=lambda x: x) diff --git a/telethon_generator/generators/errors.py b/telethon_generator/generators/errors.py index 386575be2..ad03278c4 100644 --- a/telethon_generator/generators/errors.py +++ b/telethon_generator/generators/errors.py @@ -37,8 +37,7 @@ def generate_errors(errors, f): f.write('def __init__(self, request):\n ' ' self.request = request\n ') - f.write('super(Exception, self).__init__(' - '{}'.format(repr(error.description))) + f.write(f'super(Exception, self).__init__({repr(error.description)}') if error.has_captures: f.write('.format({0}=self.{0})'.format(error.capture_name)) diff --git a/telethon_generator/generators/tlobject.py b/telethon_generator/generators/tlobject.py index 4326b189b..ab99a32dd 100644 --- a/telethon_generator/generators/tlobject.py +++ b/telethon_generator/generators/tlobject.py @@ -56,7 +56,7 @@ def _write_modules( # namespace_tlobjects: {'namespace', [TLObject]} out_dir.mkdir(parents=True, exist_ok=True) for ns, tlobjects in namespace_tlobjects.items(): - file = out_dir / '{}.py'.format(ns or '__init__') + file = out_dir / f"{ns or '__init__'}.py" with file.open('w') as f, SourceBuilder(f) as builder: builder.writeln(AUTO_GEN_NOTICE) @@ -106,8 +106,7 @@ def _write_modules( if not constructors: pass elif len(constructors) == 1: - type_defs.append('Type{} = {}'.format( - type_name, constructors[0].class_name)) + type_defs.append(f'Type{type_name} = {constructors[0].class_name}') else: type_defs.append('Type{} = Union[{}]'.format( type_name, ','.join(c.class_name @@ -124,11 +123,11 @@ def _write_modules( if not name or name in primitives: continue - import_space = '{}.tl.types'.format('.' * depth) + import_space = f"{'.' * depth}.tl.types" if '.' in name: namespace = name.split('.')[0] name = name.split('.')[1] - import_space += '.{}'.format(namespace) + import_space += f'.{namespace}' if name not in type_names: type_names.add(name) @@ -137,7 +136,7 @@ def _write_modules( continue elif import_space not in imports: imports[import_space] = set() - imports[import_space].add('Type{}'.format(name)) + imports[import_space].add(f'Type{name}') # Add imports required for type checking if imports: @@ -188,11 +187,12 @@ def _write_class_init(tlobject, kind, type_constructors, builder): builder.writeln() # Convert the args to string parameters, flags having =None - args = ['{}: {}{}'.format( - a.name, a.type_hint(), '=None' if a.is_flag or a.can_be_inferred else '') + args = [ + f"{a.name}: {a.type_hint()}{'=None' if a.is_flag or a.can_be_inferred else ''}" for a in tlobject.real_args ] + # Write the __init__ function if it has any argument if not tlobject.real_args: return @@ -224,12 +224,10 @@ def _write_class_init(tlobject, kind, type_constructors, builder): if not arg.can_be_inferred: builder.writeln('self.{0} = {0}', arg.name) - # Currently the only argument that can be - # inferred are those called 'random_id' elif arg.name == 'random_id': # Endianness doesn't really matter, and 'big' is shorter - code = "int.from_bytes(os.urandom({}), 'big', signed=True)" \ - .format(8 if arg.type == 'long' else 4) + code = f"int.from_bytes(os.urandom({8 if arg.type == 'long' else 4}), 'big', signed=True)" + if arg.is_vector: # Currently for the case of "messages.forwardMessages" @@ -239,7 +237,7 @@ def _write_class_init(tlobject, kind, type_constructors, builder): raise ValueError( 'Cannot infer list of random ids for ', tlobject ) - code = '[{} for _ in range(len(id))]'.format(code) + code = f'[{code} for _ in range(len(id))]' builder.writeln( "self.random_id = random_id if random_id " @@ -252,36 +250,40 @@ def _write_class_init(tlobject, kind, type_constructors, builder): def _write_resolve(tlobject, builder): - if tlobject.is_function and any( - (arg.type in AUTO_CASTS - or ((arg.name, arg.type) in NAMED_AUTO_CASTS - and tlobject.fullname not in NAMED_BLACKLIST)) - for arg in tlobject.real_args + if not tlobject.is_function or not any( + ( + arg.type in AUTO_CASTS + or ( + (arg.name, arg.type) in NAMED_AUTO_CASTS + and tlobject.fullname not in NAMED_BLACKLIST + ) + ) + for arg in tlobject.real_args ): - builder.writeln('async def resolve(self, client, utils):') - for arg in tlobject.real_args: - ac = AUTO_CASTS.get(arg.type) - if not ac: - ac = NAMED_AUTO_CASTS.get((arg.name, arg.type)) - if not ac: - continue + return + builder.writeln('async def resolve(self, client, utils):') + for arg in tlobject.real_args: + ac = AUTO_CASTS.get(arg.type) + if not ac: + ac = NAMED_AUTO_CASTS.get((arg.name, arg.type)) + if not ac: + continue - if arg.is_flag: - builder.writeln('if self.{}:', arg.name) + if arg.is_flag: + builder.writeln('if self.{}:', arg.name) - if arg.is_vector: - builder.writeln('_tmp = []') - builder.writeln('for _x in self.{0}:', arg.name) - builder.writeln('_tmp.append({})', ac.format('_x')) - builder.end_block() - builder.writeln('self.{} = _tmp', arg.name) - else: - builder.writeln('self.{} = {}', arg.name, - ac.format('self.' + arg.name)) + if arg.is_vector: + builder.writeln('_tmp = []') + builder.writeln('for _x in self.{0}:', arg.name) + builder.writeln('_tmp.append({})', ac.format('_x')) + builder.end_block() + builder.writeln('self.{} = _tmp', arg.name) + else: + builder.writeln('self.{} = {}', arg.name, ac.format(f'self.{arg.name}')) - if arg.is_flag: - builder.end_block() - builder.end_block() + if arg.is_flag: + builder.end_block() + builder.end_block() def _write_to_dict(tlobject, builder): @@ -299,19 +301,18 @@ def _write_to_dict(tlobject, builder): arg.name) else: builder.write('self.{}', arg.name) + elif arg.is_vector: + builder.write( + '[] if self.{0} is None else [x.to_dict() ' + 'if isinstance(x, TLObject) else x for x in self.{0}]', + arg.name + ) else: - if arg.is_vector: - builder.write( - '[] if self.{0} is None else [x.to_dict() ' - 'if isinstance(x, TLObject) else x for x in self.{0}]', - arg.name - ) - else: - builder.write( - 'self.{0}.to_dict() ' - 'if isinstance(self.{0}, TLObject) else self.{0}', - arg.name - ) + builder.write( + 'self.{0}.to_dict() ' + 'if isinstance(self.{0}, TLObject) else self.{0}', + arg.name + ) builder.writeln() builder.current_indent -= 1 @@ -362,7 +363,7 @@ def _write_from_reader(tlobject, builder): builder.writeln('@classmethod') builder.writeln('def from_reader(cls, reader):') for arg in tlobject.args: - _write_arg_read_code(builder, arg, tlobject, name='_' + arg.name) + _write_arg_read_code(builder, arg, tlobject, name=f'_{arg.name}') builder.writeln('return cls({})', ', '.join( '{0}=_{0}'.format(a.name) for a in tlobject.real_args)) @@ -464,7 +465,7 @@ def fmt_flag(flag): fmt = '(0 if {0} is None else {1})' else: fmt = '(0 if {0} is None or {0} is False else {1})' - return fmt.format('self.{}'.format(flag.name), 1 << flag.flag_index) + return fmt.format(f'self.{flag.name}', 1 << flag.flag_index) builder.write("struct.pack('{v}' else: - d[k] = re.sub( - r'([brf]?([\'"]).*\2)', - lambda m: '{}'.format(m.group(1)), - v - ) + d[k] = re.sub(r'([brf]?([\'"]).*\2)', lambda m: f'{m.group(1)}', v) KNOWN_NAMED_EXAMPLES = { @@ -122,20 +118,13 @@ def __init__(self, name, arg_type, generic_definition): # Strip the exclamation mark always to have only the name self.type = arg_type.lstrip('!') - # The type may be a flag (flags.IDX?REAL_TYPE) - # Note that 'flags' is NOT the flags name; this - # is determined by a previous argument - # However, we assume that the argument will always be called 'flags' - flag_match = re.match(r'flags.(\d+)\?([\w<>.]+)', self.type) - if flag_match: + if flag_match := re.match(r'flags.(\d+)\?([\w<>.]+)', self.type): self.is_flag = True self.flag_index = int(flag_match.group(1)) # Update the type to match the exact type, not the "flagged" one self.type = flag_match.group(2) - # Then check if the type is a Vector - vector_match = re.match(r'[Vv]ector<([\w\d.]+)>', self.type) - if vector_match: + if vector_match := re.match(r'[Vv]ector<([\w\d.]+)>', self.type): self.is_vector = True # If the type's first letter is not uppercase, then @@ -177,39 +166,35 @@ def type_hint(self): 'bytes': 'bytes', 'Bool': 'bool', 'true': 'bool', - }.get(cls, "'Type{}'".format(cls)) + }.get(cls, f"'Type{cls}'") if self.is_vector: - result = 'List[{}]'.format(result) + result = f'List[{result}]' if self.is_flag and cls != 'date': - result = 'Optional[{}]'.format(result) + result = f'Optional[{result}]' return result def real_type(self): - # Find the real type representation by updating it as required - real_type = self.type - if self.flag_indicator: - real_type = '#' - + real_type = '#' if self.flag_indicator else self.type if self.is_vector: if self.use_vector_id: - real_type = 'Vector<{}>'.format(real_type) + real_type = f'Vector<{real_type}>' else: - real_type = 'vector<{}>'.format(real_type) + real_type = f'vector<{real_type}>' if self.is_generic: - real_type = '!{}'.format(real_type) + real_type = f'!{real_type}' if self.is_flag: - real_type = 'flags.{}?{}'.format(self.flag_index, real_type) + real_type = f'flags.{self.flag_index}?{real_type}' return real_type def __str__(self): if self.generic_definition: - return '{{{}:{}}}'.format(self.name, self.real_type()) + return f'{{{self.name}:{self.real_type()}}}' else: - return '{}:{}'.format(self.name, self.real_type()) + return f'{self.name}:{self.real_type()}' def __repr__(self): return str(self).replace(':date', ':int').replace('?date', '?int') @@ -225,14 +210,15 @@ def as_example(self, f, indent=0): f.write('other_request') return - known = (KNOWN_NAMED_EXAMPLES.get((self.name, self.type)) - or KNOWN_TYPED_EXAMPLES.get(self.type) - or KNOWN_TYPED_EXAMPLES.get(SYNONYMS.get(self.type))) - if known: + if known := ( + KNOWN_NAMED_EXAMPLES.get((self.name, self.type)) + or KNOWN_TYPED_EXAMPLES.get(self.type) + or KNOWN_TYPED_EXAMPLES.get(SYNONYMS.get(self.type)) + ): f.write(known) return - assert self.omit_example() or self.cls, 'TODO handle ' + str(self) + assert self.omit_example() or self.cls, f'TODO handle {str(self)}' # Pick an interesting example if any for cls in self.cls: diff --git a/telethon_generator/parsers/tlobject/tlobject.py b/telethon_generator/parsers/tlobject/tlobject.py index da6c5f65e..a45f4a6c6 100644 --- a/telethon_generator/parsers/tlobject/tlobject.py +++ b/telethon_generator/parsers/tlobject/tlobject.py @@ -49,22 +49,18 @@ def __init__(self, fullname, object_id, args, result, WHITELISTED_MISMATCHING_IDS.get(layer, set()) if self.fullname not in whitelist: - assert self.id == self.infer_id(),\ - 'Invalid inferred ID for ' + repr(self) + assert self.id == self.infer_id(), f'Invalid inferred ID for {repr(self)}' self.class_name = snake_to_camel_case( self.name, suffix='Request' if self.is_function else '') - self.real_args = list(a for a in self.sorted_args() if not - (a.flag_indicator or a.generic_definition)) + self.real_args = [a for a in self.sorted_args() if not + (a.flag_indicator or a.generic_definition)] @property def innermost_result(self): index = self.result.find('<') - if index == -1: - return self.result - else: - return self.result[index + 1:-1] + return self.result if index == -1 else self.result[index + 1:-1] def sorted_args(self): """Returns the arguments properly sorted and ready to plug-in @@ -75,17 +71,9 @@ def sorted_args(self): key=lambda x: x.is_flag or x.can_be_inferred) def __repr__(self, ignore_id=False): - if self.id is None or ignore_id: - hex_id = '' - else: - hex_id = '#{:08x}'.format(self.id) - - if self.args: - args = ' ' + ' '.join([repr(arg) for arg in self.args]) - else: - args = '' - - return '{}{}{} = {}'.format(self.fullname, hex_id, args, self.result) + hex_id = '' if self.id is None or ignore_id else '#{:08x}'.format(self.id) + args = ' ' + ' '.join([repr(arg) for arg in self.args]) if self.args else '' + return f'{self.fullname}{hex_id}{args} = {self.result}' def infer_id(self): representation = self.__repr__(ignore_id=True)