util.py

import struct
from functools import wraps
import logging
import traceback

from telethon import types
from telethon.types import struct
from telethon.utils import _encode_telegram_base64, _rle_encode


# Updated version from https://github.com/LonamiWebs/Telethon/pull/3255
def pack_bot_file_id(file):
  """
  Inverse operation for `resolve_bot_file_id`.

  The only parameters this method will accept are :tl:`Document` and
  :tl:`Photo`, and it will return a variable-length ``file_id`` string.

  If an invalid parameter is given, it will ``return None``.
  """
  if isinstance(file, types.MessageMediaDocument):
    file = file.document
  elif isinstance(file, types.MessageMediaPhoto):
    file = file.photo

  if isinstance(file, types.Document):
    file_type = 5
    for attribute in file.attributes:
      if isinstance(attribute, types.DocumentAttributeAudio):
        file_type = 3 if attribute.voice else 9
      elif isinstance(attribute, types.DocumentAttributeVideo):
        file_type = 13 if attribute.round_message else 4
      elif isinstance(attribute, types.DocumentAttributeSticker):
        file_type = 8
      elif isinstance(attribute, types.DocumentAttributeAnimated):
        file_type = 10
      else:
        continue
      break

    return _encode_telegram_base64(_rle_encode(struct.pack(
      '<iiqqb', file_type, file.dc_id, file.id, file.access_hash, 2)))

  elif isinstance(file, types.Photo):
    size = next((x for x in reversed(file.sizes) if isinstance(
      x, (types.PhotoSize, types.PhotoCachedSize))), None)

    if not size:
      return None

    # Location is deprecated since layer 128. See #3242
    return _encode_telegram_base64(_rle_encode(struct.pack(
      '<iiqqqqib', 2, file.dc_id, file.id, file.access_hash,
      0, 0, 0, 2  # 0 = old `secret`
    )))
  else:
    return None


def format_exception(e):
  exc_desc_lines = traceback.format_exception_only(type(e), e)
  return ''.join(exc_desc_lines).rstrip()


def suppress_exceptions(logger: logging.Logger, *exceptions):
  def wrapper(func):
    @wraps(func)
    async def wrapped(*args, **kwargs):
      try:
        return await func(*args, **kwargs)
      except exceptions as e:
        name = getattr(func, '__name__', repr(func))
        logger.warn(f'{name}: suppressed {format_exception(e)}')

    return wrapped
  return wrapper


# Translation tables to turn ascii hex bytes into byte integer values
tt_to_hex = b'0123456789ABCDEF'.ljust(256, b'\x00')
_from_hex = {ord(c): int(c, 16) for c in '0123456789ABCDEFabcdef'}
tt_from_hex = bytes(_from_hex.get(i, 0) for i in range(256))