import json import re import urllib.request from collections import defaultdict URL = 'https://rpc.pwrtelegram.xyz/?all' 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(output, url=URL): 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( '\n\nclass {}({}):\n def __init__(self, **kwargs):\n ' ''.format(name, get_class_name(code)) ) if capture_name: f.write( "self.{} = int(kwargs.get('capture', 0))\n ".format(capture_name) ) f.write('super(Exception, self).__init__(self, {}'.format(repr(desc))) if capture_name: f.write('.format(self.{})'.format(capture_name)) f.write(')\n') def generate_code(output, json_file, errors_desc): 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].update(( 'FLOOD_WAIT_X', 'FLOOD_TEST_PHONE_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(errors_desc, 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', 'FloodTestPhoneWaitError': '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( 'from .rpc_base_errors import RPCError, BadMessageError, {}\n'.format( ", ".join(known_base_classes.values())) ) for code, cls in needed_base_classes: f.write( '\n\nclass {}(RPCError):\n code = {}\n'.format(cls, code) ) 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 across different files? write_error(f, error_code, name, description, capture) f.write('\n\nrpc_errors_all = {\n') for pattern, name in patterns: f.write(' {}: {},\n'.format(repr(pattern), name)) f.write('}\n')