poyee-data-warehouse/dw_base/utils/config_utils.py

from typing import Tuple, Dict, Union, List

from dw_base import NORM_GRN, NORM_MGT
from dw_base.utils.log_utils import pretty_print


def parse_args(args: [str]) -> Tuple[Dict[str, Union[str, bool, List[str]]], list]:
    """
    解析参数，返回参数map和未识别参数
    Args:
        args: 参数
    Returns: 参数map和未识别参数
    """
    args_length = len(args)
    unrecognized_args = []
    config = {}
    skip_next = False
    for index in range(0, args_length):
        if skip_next:
            skip_next = False
            continue
        arg = args[index]
        if arg.startswith('--') or arg.startswith('-'):
            if arg.startswith('--'):
                key_start = 2
            else:
                key_start = 1
            if arg.__contains__('='):
                # --key=value形式或-key=value形式
                key = arg[key_start:arg.index('=')]
                value = arg[arg.index('=') + 1:].strip()
            elif index + 1 == args_length:
                # --key形式或-key形式，且是最后一个参数
                key = arg[key_start:]
                value = True
            else:
                # --key形式或-key形式，后面还有参数
                key = arg[key_start:]
                next_arg = args[index + 1]
                if next_arg.startswith('--') or next_arg.startswith('-'):
                    # 下一个参数也是--key或-key形式
                    value = True
                else:
                    # 下一个参数是当前key的value，跳过下一个参数的处理
                    value = next_arg.strip()
                    skip_next = True
            if config.__contains__(key):
                # 列表类配置，例如: -c c1 -c c2 -c c3
                if isinstance(config[key], list):
                    # 第三次以后识别到
                    list_value = config.get(key)  # type:List
                    if not list_value.__contains__(value):
                        list_value.append(value)
                        config[key] = list_value
                else:
                    # 第二次识别到
                    if config[key] != value:
                        config[key] = [config.get(key), value]
            else:
                # 第一次识别到
                config[key] = value
        else:
            # 其他情况的参数丢弃
            unrecognized_args.append(arg)
    print_config(config)
    return config, unrecognized_args


def print_config(config: Dict[str, str]):
    """
    打印配置
    Args:
        config: 配置
    Returns:
    """
    for key, value in config.items():
        if isinstance(value, list):
            key_value = f'{NORM_GRN}{key} => [{", ".join(value)}]'
        else:
            key_value = f'{NORM_GRN}{key} => {str(value)}'
        pretty_print(f'{NORM_MGT}解析到参数： {key_value}')