Generate errors from PWRTelegram's API

telethon_generator/error_generator.py

@@ -0,0 +1,172 @@
+import json
+import re
+import urllib.request
+from collections import defaultdict
+
+URL = 'https://rpc.pwrtelegram.xyz/?all'
+OUTPUT = '../telethon/errors/rpc_error_list.py'
+JSON_OUTPUT = 'errors.json'
+
+known_base_classes = {
+    303: 'InvalidDCError',
+    400: 'BadRequestError',
+    401: 'UnauthorizedError',
+    403: 'ForbiddenError',
+    404: 'NotFoundError',
+    420: 'FloodError',
+    500: 'ServerError',
+}
+
+# The API doesn't return the code for some (vital) errors. They are
+# all assumed to be 400, except these well-known ones that aren't.
+known_codes = {
+    'ACTIVE_USER_REQUIRED': 401,
+    'AUTH_KEY_UNREGISTERED': 401,
+    'USER_DEACTIVATED': 401
+}
+
+
+def fetch_errors(url=URL, output=JSON_OUTPUT):
+    print('Opening a connection to', url, '...')
+    r = urllib.request.urlopen(url)
+    print('Checking response...')
+    data = json.loads(
+        r.read().decode(r.info().get_param('charset') or 'utf-8')
+    )
+    if data.get('ok'):
+        print('Response was okay, saving data')
+        with open(output, 'w', encoding='utf-8') as f:
+            json.dump(data, f)
+        return True
+    else:
+        print('The data received was not okay:')
+        print(json.dumps(data, indent=4))
+        return False
+
+
+def get_class_name(error_code):
+    if isinstance(error_code, int):
+        return known_base_classes.get(
+            error_code, 'RPCError' + str(error_code).replace('-', 'Neg')
+        )
+
+    if 'FIRSTNAME' in error_code:
+        error_code = error_code.replace('FIRSTNAME', 'FIRST_NAME')
+
+    result = re.sub(
+        r'_([a-z])', lambda m: m.group(1).upper(), error_code.lower()
+    )
+    return result[:1].upper() + result[1:].replace('_', '') + 'Error'
+
+
+def write_error(f, code, name, desc, capture_name):
+    f.write(
+        f'\n'
+        f'\n'
+        f'class {name}({get_class_name(code)}):\n'
+        f'    def __init__(self, **kwargs):\n'
+        f'        '
+    )
+    if capture_name:
+        f.write(
+            f"self.{capture_name} = int(kwargs.get('capture', 0))\n"
+            f"        "
+        )
+    f.write(f'super(Exception, self).__init__(self, {repr(desc)}')
+    if capture_name:
+        f.write(f'.format(self.{capture_name})')
+    f.write(')\n')
+
+
+def generate_code(json_file=JSON_OUTPUT, output=OUTPUT):
+    with open(json_file, encoding='utf-8') as f:
+        data = json.load(f)
+
+    errors = defaultdict(set)
+    # PWRTelegram's API doesn't return all errors, which we do need here.
+    # Add some special known-cases manually first.
+    errors[420].add('FLOOD_WAIT_X')
+    errors[401].update((
+        'AUTH_KEY_INVALID', 'SESSION_EXPIRED', 'SESSION_REVOKED'
+    ))
+    errors[303].update((
+        'FILE_MIGRATE_X', 'PHONE_MIGRATE_X',
+        'NETWORK_MIGRATE_X', 'USER_MIGRATE_X'
+    ))
+    for error_code, method_errors in data['result'].items():
+        for error_list in method_errors.values():
+            for error in error_list:
+                errors[int(error_code)].add(re.sub('_\d+', '_X', error).upper())
+
+    # Some errors are in the human result, but not with a code. Assume code 400
+    for error in data['human_result']:
+        if error[0] != '-' and not error.isdigit():
+            error = re.sub('_\d+', '_X', error).upper()
+            if not any(error in es for es in errors.values()):
+                errors[known_codes.get(error, 400)].add(error)
+
+    # Some error codes are not known, so create custom base classes if needed
+    needed_base_classes = [
+        (e, get_class_name(e)) for e in errors if e not in known_base_classes
+    ]
+
+    # Prefer the descriptions that are related with Telethon way of coding to
+    # those that PWRTelegram's API provides.
+    telethon_descriptions = {}
+    with open('error_descriptions', encoding='utf-8') as f:
+        for line in f:
+            line = line.strip()
+            if line and not line.startswith('#'):
+                equal = line.index('=')
+                message, description = line[:equal], line[equal + 1:]
+                telethon_descriptions[message.rstrip()] = description.lstrip()
+
+    # Names for the captures, or 'x' if unknown
+    capture_names = {
+        'FloodWaitError': 'seconds',
+        'FileMigrateError': 'new_dc',
+        'NetworkMigrateError': 'new_dc',
+        'PhoneMigrateError': 'new_dc',
+        'UserMigrateError': 'new_dc',
+        'FilePartMissingError': 'which'
+    }
+
+    # Everything ready, generate the code
+    with open(output, 'w', encoding='utf-8') as f:
+        f.write(
+            f'from .rpc_base_errors import RPCError, BadMessageError, '
+            f'{", ".join(known_base_classes.values())}\n'
+        )
+        for code, cls in needed_base_classes:
+            f.write(
+                f'\n'
+                f'\n'
+                f'class {cls}(RPCError):\n'
+                f'    code = {code}\n'
+            )
+
+        patterns = []  # Save this dictionary later in the generated code
+        for error_code, error_set in errors.items():
+            for error in sorted(error_set):
+                description = telethon_descriptions.get(
+                    error, '\n'.join(data['human_result'].get(
+                        error, ['No description known.']
+                    ))
+                )
+                has_captures = '_X' in error
+                if has_captures:
+                    name = get_class_name(error.replace('_X', ''))
+                    pattern = error.replace('_X', r'_(\d+)')
+                else:
+                    name, pattern = get_class_name(error), error
+
+                patterns.append((pattern, name))
+                capture = capture_names.get(name, 'x') if has_captures else None
+                # TODO Some errors have the same name but different code,
+                # split this accross different files?
+                write_error(f, error_code, name, description, capture)
+
+        f.write('\n\nrpc_errors_all = {\n')
+        for pattern, name in patterns:
+            f.write(f'    {repr(pattern)}: {name},\n')
+        f.write('}\n')