Telethon/telethon/errors/__init__.py

"""
This module holds all the base and automatically generated errors that the
Telegram API has. See telethon_generator/errors.json for more.
"""
import urllib.request
import re
from threading import Thread

from .common import (
    ReadCancelledError, TypeNotFoundError, InvalidChecksumError,
    BrokenAuthKeyError, SecurityError, CdnFileTamperedError, MultiError
)

# This imports the base errors too, as they're imported there
from .rpcbaseerrors import *
from .rpcerrorlist import *


def report_error(code, message, report_method):
    """
    Reports an RPC error to pwrtelegram.

    :param code: the integer code of the error (like 400).
    :param message: the message representing the error.
    :param report_method: the constructor ID of the function that caused it.
    """
    try:
        # Ensure it's signed
        report_method = int.from_bytes(
            report_method.to_bytes(4, 'big'), 'big', signed=True
        )
        url = urllib.request.urlopen(
            'https://rpc.pwrtelegram.xyz?code={}&error={}&method={}'
            .format(code, message, report_method),
            timeout=5
        )
        url.read()
        url.close()
    except Exception as e:
        "We really don't want to crash when just reporting an error"


def rpc_message_to_error(rpc_error, report_method=None):
    """
    Converts a Telegram's RPC Error to a Python error.

    :param rpc_error: the RpcError instance.
    :param report_method: if present, the ID of the method that caused it.
    :return: the RPCError as a Python exception that represents this error.
    """
    if report_method is not None:
        Thread(
            target=report_error,
            args=(rpc_error.error_code, rpc_error.error_message, report_method)
        ).start()

    # Try to get the error by direct look-up, otherwise regex
    cls = rpc_errors_dict.get(rpc_error.error_message, None)
    if cls:
        return cls()

    for msg_regex, cls in rpc_errors_re:
        m = re.match(msg_regex, rpc_error.error_message)
        if m:
            capture = int(m.group(1)) if m.groups() else None
            return cls(capture=capture)

    cls = base_errors.get(rpc_error.error_code)
    if cls:
        return cls(rpc_error.error_message)

    return RPCError(rpc_error.error_message, rpc_error.error_code)