lbry-sdk/lbry/service/parser.py

import json
import inspect
import textwrap
import tokenize
import token
from io import BytesIO
from typing import Tuple, List

from lbry.service import api
from lbry.service import json_encoder


LINE_WIDTH = 90


def parse_description(desc) -> dict:
    lines = iter(desc.splitlines())
    parts = {'text': []}
    current = parts['text']
    for line in lines:
        if line.strip() in ('Usage:', 'Options:', 'Returns:'):
            current = parts.setdefault(line.strip().lower()[:-1], [])
        else:
            if line.strip():
                if line.strip() == '{kwargs}':
                    parts['kwargs'] = line.find('{kwargs}')
                else:
                    current.append(line)
    return parts


def parse_type(tokens: List) -> Tuple[str, str]:
    type_ = [tokens.pop(0).string]
    if tokens and tokens[0].string == '[':
        while tokens[0].string != ']':
            type_.append(tokens.pop(0).string)
        type_.append(tokens.pop(0).string)
    json_ = None
    if type_ == ['StrOrList']:
        type_ = ['str, list']
    elif type_[0] == 'Paginated':
        json_ = json_encoder.encode_pagination_doc(
            getattr(json_encoder, f'{type_[2].lower()}_doc')
        )
    elif len(type_) == 1 and hasattr(json_encoder, f'{type_[0].lower()}_doc'):
        json_ = getattr(json_encoder, f'{type_[0].lower()}_doc')
    return ''.join(type_), json_


def parse_argument(tokens, method_name='') -> dict:
    arg = {
        'name': tokens.pop(0).string,
        'desc': []
    }
    if arg['name'] == 'self':
        return {}
    if tokens[0].string == ':':
        tokens.pop(0)
        type_tokens = []
        while tokens[0].string not in ('=', ',', ')') and tokens[0].type != token.COMMENT:
            type_tokens.append(tokens.pop(0))
        arg['type'] = parse_type(type_tokens)[0]
    if tokens[0].string == '=':
        tokens.pop(0)
        default = tokens.pop(0)
        if default.type == token.NAME:
            default_value = eval(default.string)  # pylint: disable=eval-used
            if default_value is not None:
                arg['default'] = default_value
        elif default.type == token.NUMBER:
            arg['default'] = int(default.string)
        else:
            arg['default'] = default.string
    if tokens[0].string == ',':
        tokens.pop(0)
    if 'type' not in arg:
        if 'default' in arg:
            arg['type'] = type(arg['default']).__name__
        else:
            raise Exception(f"No type and no default value for {arg['name']} in {method_name}")
    for part in tokens:
        if part.type == token.COMMENT:
            arg['desc'].append(part.string[1:].strip())
    return arg


def produce_argument_tokens(src: str):
    in_args = False
    in_comment = False
    parsed = []
    for t in tokenize.tokenize(BytesIO(src.encode()).readline):
        if t.string == '(':
            in_args = True
        elif in_args:
            if not in_comment and t.string == ',':
                in_comment = True
            elif in_comment and (t.type == token.NAME or t.string == '**'):
                if not parsed[0].string.startswith('_'):
                    yield parsed
                in_comment = False
                parsed = []
            if t.type in (token.NAME, token.OP, token.COMMENT, token.STRING, token.NUMBER):
                parsed.append(t)
            if t.string == ')':
                if not parsed[0].string.startswith('_'):
                    yield parsed
                break


def parse_return(tokens) -> dict:
    d = {'desc': [], 'type': None}
    if tokens[0].string == '->':
        tokens.pop(0)
        type_tokens = []
        while tokens[0].string != ':':
            type_tokens.append(tokens.pop(0))
        d['type'], _ = parse_type(type_tokens)
        if _:
            d['json'] = _
    assert tokens.pop(0).string == ':'
    for part in tokens:
        if part.type == token.COMMENT:
            d['desc'].append(part.string[1:].strip())
    return d


def produce_return_tokens(src: str):
    in_return = False
    parsed = []
    for t in tokenize.tokenize(BytesIO(src.encode()).readline):
        if t.string == ')':
            in_return = True
        elif in_return:
            if t.type == token.INDENT:
                break
            parsed.append(t)
    return parsed


def parse_method(method, expanders: dict) -> dict:
    d = {
        'name': method.__name__,
        'desc': parse_description(textwrap.dedent(method.__doc__)) if method.__doc__ else {},
        'method': method,
        'arguments': [],
        'returns': None
    }
    src = inspect.getsource(method)
    for tokens in produce_argument_tokens(src):
        if tokens[0].string == '**':
            tokens.pop(0)
            d['kwargs'] = []
            expander_names = tokens.pop(0).string[:-7]
            if expander_names.startswith('_'):
                continue
            for expander_name in expander_names.split('_and_'):
                if expander_name not in expanders:
                    raise Exception(f"Expander '{expander_name}' not found, used by {d['name']}.")
                d['arguments'].extend(expanders[expander_name])
                d['kwargs'].extend(expanders[expander_name])
        else:
            arg = parse_argument(tokens, d['name'])
            if arg:
                d['arguments'].append(arg)
    d['returns'] = parse_return(produce_return_tokens(src))
    return d


def get_expanders():
    expanders = {}
    for name, func in api.kwarg_expanders.items():
        if name.endswith('_original'):
            expanders[name[:-len('_original')]] = parse_method(func, expanders)['arguments']
    return expanders


def get_groups(cls):
    return {
        group_name[:-len('_DOC')].lower(): getattr(cls, group_name).strip()
        for group_name in dir(cls) if group_name.endswith('_DOC')
    }


def get_methods(cls):
    expanders = get_expanders()
    return {
        method: parse_method(getattr(cls, method), expanders)
        for method in dir(cls) if not method.endswith('_DOC') and not method.startswith('_')
    }


def generate_options(method, indent) -> List[str]:
    if not method['arguments']:
        return []
    flags = []
    for arg in method['arguments']:
        if arg['type'] == 'bool':
            flags.append(f"--{arg['name']}")
        else:
            flags.append(f"--{arg['name']}=<{arg['name']}>")
    max_len = max(len(f) for f in flags) + 2
    flags = [f.ljust(max_len) for f in flags]
    options = []
    for flag, arg in zip(flags, method['arguments']):
        left = f"{indent}{flag}: "
        text = f"({arg['type']}) {' '.join(arg['desc'])}"
        if 'default' in arg:
            if arg['type'] != 'bool':
                text += f" [default: {arg['default']}]"
        wrapped = textwrap.wrap(text, LINE_WIDTH-len(left))
        lines = [f"{left}{wrapped.pop(0)}"]
        for line in wrapped:
            lines.append(f"{' '*len(left)} {line}")
        options.extend(lines)
    return options


def generate_help(command):
    indent = 4
    text = []
    desc = command['desc']

    for line in desc.get('text', []):
        text.append(line)
    text.append('')

    usage, kwargs_offset = desc.get('usage', []), desc.get('kwargs', False)
    text.append('Usage:')
    if usage:
        for line in usage:
            text.append(line)
    else:
        text.append(f"{' '*indent}{command['cli']}")
    if kwargs_offset:
        flags = []
        for arg in command['kwargs']:
            if arg['type'] == 'bool':
                flags.append(f"[--{arg['name']}]")
            elif 'list' in arg['type']:
                flags.append(f"[--{arg['name']}=<{arg['name']}>...]")
            else:
                flags.append(f"[--{arg['name']}=<{arg['name']}>]")
        wrapped = textwrap.wrap(' '.join(flags), LINE_WIDTH-kwargs_offset)
        for line in wrapped:
            text.append(f"{' '*kwargs_offset}{line}")
    text.append('')

    options = desc.get('options', [])
    if options or command['arguments']:
        text.append('Options:')
        for line in options:
            text.append(line)
        text.extend(generate_options(command, ' '*indent))
        text.append('')

    returns = desc.get('returns', [])
    if returns or command['returns']['type']:
        text.append('Returns:')
        if command['returns']['type']:
            return_comment = ' '.join(command['returns']['desc'])
            text.append(f"{' '*indent}({command['returns']['type']}) {return_comment}")
        text.extend(returns)
        if 'json' in command['returns']:
            dump = json.dumps(command['returns']['json'], indent=4)
            text.extend(textwrap.indent(dump, ' '*indent).splitlines())

    return '\n'.join(text)


def get_api_definitions(cls):
    groups = get_groups(cls)
    commands = get_methods(cls)
    for name, command in commands.items():
        parts = name.split('_')
        if parts[0] in groups:
            command['name'] = '_'.join(parts[1:])
            command['group'] = parts[0]
            command['cli'] = f"{command['group']} {command['name']}"
        else:
            command['cli'] = command['name']
        command['help'] = generate_help(command)
    return {'groups': groups, 'commands': commands}


def write(fp):
    fp.write('# pylint: skip-file\n')
    fp.write('# DO NOT EDIT: GENERATED FILE\n')
    fp.write('interface = ')
    defs = get_api_definitions(api.API)
    for c in defs['commands'].values():
        del c['method']
    j = json.dumps(defs, indent=4)
    j = j.replace(': false', ': False')
    j = j.replace(': true', ': True')
    j = j.replace(': null', ': None')
    fp.write(j)
    fp.write('\n')


def main():
    with open('metadata.py', 'w') as fp:
        write(fp)


if __name__ == "__main__":
    main()