PyWebIO/wsrepl/interact.py

import json
import logging
from collections.abc import Mapping

from .framework import Global
from .input_ctrl import send_msg, single_input, input_control

logger = logging.getLogger(__name__)


def run_async(coro):
    Global.active_ws.inactive_coro_instances.append(coro)


TEXT = 'text'
NUMBER = "number"
PASSWORD = "password"
CHECKBOX = 'checkbox'
RADIO = 'radio'
SELECT = 'select'
TEXTAREA = 'textarea'


def _parse_args(kwargs):
    """处理传给各类input函数的原始参数，
    :return:（spec参数，valid_func）
    """
    # 对为None的参数忽律处理
    kwargs = {k: v for k, v in kwargs.items() if v is not None}
    kwargs.update(kwargs.get('other_html_attrs', {}))
    kwargs.pop('other_html_attrs', None)
    valid_func = kwargs.pop('valid_func', lambda _: None)
    return kwargs, valid_func


def input(label, type=TEXT, *, valid_func=None, name='data', value=None, placeholder=None, required=None,
          readonly=None, disabled=None, **other_html_attrs):
    item_spec, valid_func = _parse_args(locals())

    # 参数检查
    allowed_type = {TEXT, NUMBER, PASSWORD, TEXTAREA}
    assert type in allowed_type, 'Input type not allowed.'

    def preprocess_func(d):
        if type == NUMBER:
            return int(d)
        return d

    return single_input(item_spec, valid_func, preprocess_func)


def textarea(label, rows=6, *, code=None, valid_func=None, name='data', value=None, placeholder=None, required=None,
             maxlength=None, minlength=None, readonly=None, disabled=None, **other_html_attrs):
    item_spec, valid_func = _parse_args(locals())
    item_spec['type'] = TEXTAREA

    return single_input(item_spec, valid_func, lambda d: d)


def _parse_select_options(options):
    # option 可用形式：
    # {value:, label:, [selected:,] [disabled:]}
    # (value, label, [selected,] [disabled])
    # value 单值，label等于value
    opts_res = []
    for opt in options:
        if isinstance(opt, Mapping):
            assert 'value' in opt and 'label' in opt, 'options item must have value and label key'
        elif isinstance(opt, list):
            assert len(opt) > 1 and len(opt) <= 4, 'options item format error'
            opt = dict(zip(('value', 'label', 'selected', 'disabled'), opt))
        else:
            opt = dict(value=opt, label=opt)
        opts_res.append(opt)

    return opts_res


def select(label, options, type=SELECT, *, multiple=None, valid_func=None, name='data', value=None,
           placeholder=None, required=None, readonly=None, disabled=None, inline=None, **other_html_attrs):
    """
    参数值为None表示不指定，使用默认值

    :param label:
    :param options: option 列表
        option 可用形式：
        {value:, label:, [selected:,] [disabled:]}
        (value, label, [selected,] [disabled])
        value 单值，label等于value
    :param type:
    :param multiple:
    :param valid_func:
    :param name:
    :param value:
    :param placeholder:
    :param required:
    :param readonly:
    :param disabled:
    :param inline:
    :param other_html_attrs:
    :return:
    """
    item_spec, valid_func = _parse_args(locals())
    item_spec['options'] = _parse_select_options(options)

    allowed_type = {CHECKBOX, RADIO, SELECT}
    assert type in allowed_type, 'Input type not allowed.'

    if inline is not None and type not in {CHECKBOX, RADIO}:
        del item_spec['inline']
        logger.warning('inline 只能用于 CHECKBOX, RADIO type, now type:%s', type)

    if multiple is not None and type != SELECT:
        del item_spec['multiple']
        logger.warning('multiple 参数只能用于SELECT type, now type:%s', type)

    return single_input(item_spec, valid_func, lambda d: d)


def _parse_action_buttons(buttons):
    """
    :param label:
    :param actions: action 列表
    action 可用形式：
        {value:, label:, [disabled:]}
        (value, label, [disabled])
        value 单值，label等于value
    :return:
    """
    act_res = []
    for act in buttons:
        if isinstance(act, Mapping):
            assert 'value' in act and 'label' in act, 'actions item must have value and label key'
        elif isinstance(act, list):
            assert len(act) in (2, 3), 'actions item format error'
            act = dict(zip(('value', 'label', 'disabled'), act))
        else:
            act = dict(value=act, label=act)
        act_res.append(act)

    return act_res


def actions(label, buttons, name='data'):
    """
    选择一个动作。UI为多个按钮，点击后会将整个表单提交
    :param label:
    :param actions: action 列表
    action 可用形式：
        {value:, label:, [disabled:]}
        (value, label, [disabled])
        value 单值，label等于value

    实现方式：
    多个type=submit的input组成
    [
        <button data-name value>label</button>,
        ...
    ]
    """
    item_spec, valid_func = _parse_args(locals())
    item_spec['type'] = 'actions'
    item_spec['buttons'] = _parse_action_buttons(buttons)

    return single_input(item_spec, valid_func, lambda d: d)


def set_title(title):
    send_msg('output_ctl', dict(title=title))


def text_print(text, *, ws=None):
    msg = dict(command="output", spec=dict(content=text, type='text'))
    (ws or Global.active_ws).write_message(json.dumps(msg))


def json_print(obj):
    text = "```\n%s\n```" % json.dumps(obj, indent=4, ensure_ascii=False)
    text_print(text)


put_markdown = text_print


def input_group(label, inputs, valid_func=None):
    """
    :param label:
    :param inputs: list of single_input coro
    :param valid_func: callback(data) -> (name, error_msg)
    :return:
    """
    spec_inputs = []
    preprocess_funcs = {}
    item_valid_funcs = {}
    for single_input_cr in inputs:
        input_kwargs = dict(single_input_cr.cr_frame.f_locals)  # 拷贝一份，不可以对locals进行修改
        single_input_cr.close()
        input_name = input_kwargs['item_spec']['name']
        preprocess_funcs[input_name] = input_kwargs['preprocess_func']
        item_valid_funcs[input_name] = input_kwargs['valid_func']
        spec_inputs.append(input_kwargs['item_spec'])

    # def add_autofocus(spec_inputs):
    if all('autofocus' not in i for i in spec_inputs):  # 每一个输入项都没有设置autofocus参数
        for i in spec_inputs:
            text_inputs = {TEXT, NUMBER, PASSWORD, SELECT}  # todo update
            if i.get('type') in text_inputs:
                i['autofocus'] = True
                break

    spec = dict(label=label, inputs=spec_inputs)
    return input_control(spec, preprocess_funcs=preprocess_funcs, item_valid_funcs=item_valid_funcs,
                         form_valid_funcs=valid_func)