d3i-szct/cli.py

"""
控制台工具。

包含 Web 服务管理、远程数据获取服务管理、数据导入命令、用户管理命令、权限设置命令等。
可直接在控制台终端实现部分数据操作与管理。
"""

import argparse
import datetime
import logging
import os.path
import re
import sys
from typing import Optional, Union

from dateutil.relativedelta import relativedelta

import apps
from dock.dcm import dcm_security
from dock.govs import govs_security
from dock.oa import oa_security
from paste.core import aio_pool
from paste.core.logging import echo_log
from paste.db import gen_models
from paste.security import token
from paste.service import server
from paste.util import udict

ArgParser = argparse.ArgumentParser(
    prefix_chars='-+',
    description=f"命令行管理工具。",
)
"""
命令行参数解析器。
"""

ArgSubparsers = ArgParser.add_subparsers(
    dest='sub_command',
    title=f'内部命令',
    description=f'请使用以下命令以及对应选项完成具体任务。',
    help='使用 python3 cli.py [command] -h 查看使用说明。'
)
"""
内部命令。
"""

ArgNamespace: Optional[argparse.Namespace] = None
"""
命令行参数解析结果。
"""

HadBehavior = False
"""
是否正确执行了行为函数。
"""


def had_behavior(success: Optional[bool] = None):
    """
    是否成功执行了行为方法，可取得或设置执行状态。

    :param success: 执行状态
    :return: 执行状态
    """
    global HadBehavior
    if success is not None:
        HadBehavior = success
    return HadBehavior


def current_dir():
    print(f"当前目录：{os.path.abspath(os.path.curdir)}")


def version():
    print(apps.get_version())


def generate_models():
    """
    重建数据模型。
    """
    try:
        _runner = aio_pool.get_aio_runner()
        _runner(gen_models.sqlacodegen())
        echo_log(msg='成功重建数据模型')
    except Exception as e:
        echo_log(msg=e, level=logging.ERROR, is_log_exc=True)


def generate_token(client_id: str = 'Any'):
    """
    生成一个长期 Token 默认 50 年。

    :params client_id: 任意
    """
    try:
        _now = datetime.datetime.now(datetime.timezone.utc)
        _exp = _now + relativedelta(years=50)
        _token = token.encode_token(exp=_exp, client_id=client_id)
        echo_log(msg=f"过期时间：{_exp}")
        echo_log(msg=_token)
    except Exception as e:
        echo_log(msg=e, level=logging.ERROR, is_log_exc=True)


def renew_token():
    _runner = aio_pool.get_aio_runner()
    try:
        echo_log(f"开始执行数字城管 Cookies 更新...")
        _runner(dcm_security.login())
        echo_log(f"完成数字城管 Cookies 更新.")
    except Exception as e:
        echo_log(msg=e, level=logging.ERROR, is_log_exc=True)

    try:
        echo_log(f"开始执行 OA Token 更新...")
        _runner(oa_security.login())
        echo_log(f"完成 OA Token 更新.")
    except Exception as e:
        echo_log(msg=e, level=logging.ERROR, is_log_exc=True)

    try:
        echo_log(f"开始执行省12345 Token 更新...")
        _runner(govs_security.login())
        echo_log(f"完成省12345 Token 更新.")
    except Exception as e:
        echo_log(msg=e, level=logging.ERROR, is_log_exc=True)


def push_task(task_id: Union[str, int]):
    """
    单条推送测试。

    :param task_id: 待办任务 ID
    """
    async def _push_task(_task_id: Union[str, int]):
        from dock.oa_dcm.oa_push_order import push_order
        await push_order(task_id=_task_id)

        from dock.oa_dcm.oa_push_order_detail import push_order_detail
        await push_order_detail(task_id=_task_id)

        from dock.oa_dcm.oa_push_extend_info import push_extend_info
        await push_extend_info(task_id=_task_id)

        from dock.oa_dcm.oa_push_more_info import push_more_info
        await push_more_info(task_id=_task_id)

        from dock.oa_dcm.oa_upload import upload_with_attachment
        await upload_with_attachment(task_id=_task_id)

        from dock.oa_dcm.oa_push_attachment import push_attachment
        await push_attachment(task_id=_task_id)

        from dock.oa_dcm.oa_push_process_info import push_process_info
        await push_process_info(task_id=_task_id)

        # 推送结束后，签收这些已推送的工单
        from dock.oa_dcm.oa_sign_task import sign_task
        await sign_task(task_id=_task_id)

    _runner = aio_pool.get_aio_runner()
    _runner(_push_task(task_id))


def start_service(full_service_name):
    """
    在控制台启动服务，当控制台关闭时，服务停止。

    :param full_service_name: 完整服务名称：package.module
    """
    server.start_service(full_service_name)


def start(full_service_name):
    """
    启动常驻内存服务。

    :param full_service_name: 完整服务名称：package.module
    """
    server.start(full_service_name)


def stop(full_service_name):
    """
    停止常驻内存服务。

    :param full_service_name: 完整服务名称：package.module
    """
    server.stop(full_service_name)


tool_config = {
    'cmds': {
        'generate_models': {
            'behavior': generate_models,
            'args': {},
            'opts': {
                'help': f"重建（覆盖）数据模型，注意：第一、用户模型与 RbacUserModel 模型冲突，须删除，注意其外键引用问题。"
            },
        },
        'generate_token': {
            'behavior': generate_token,
            'args': {
                'client_id': {'action': 'store', 'nargs': '?', 'help': '如: backend.0ad8d20fbd2804c038f19c9f522010d3'},
            },
            'opts': {
                'help': f"创建长期 Token。"
            },
        },
        'renew_token': {
            'behavior': renew_token,
            'args': {},
            'opts': {
                'help': f"重建Token。"
            },
        },
        'push_task': {
            'behavior': push_task,
            'args': {
                'task_id': {'action': 'store', 'help': '如: 2054531648254513152'},
            },
            'opts': {
                'help': f"推送待办工单。"
            },
        },

        'start_service': {
            'behavior': start_service,
            'args': {
                'full_service_name': {'action': 'store', 'help': '如 package.module 格式的服务名称'},
            },
            'opts': {
                'help': f"在控制台启动服务，注意：当控制台关闭时，服务随即停止"
            },
        },
        'start': {
            'behavior': start,
            'args': {
                'full_service_name': {'action': 'store', 'help': '如 package.module 格式的服务名称'},
            },
            'opts': {
                'help': f"常驻内存方式启动服务，当控制台关闭时，服务将继续运行，直到调用 stop 命令"
            },
        },
        'stop': {
            'behavior': stop,
            'args': {
                'full_service_name': {'action': 'store', 'help': '如 package.module 格式的服务名称'},
            },
            'opts': {
                'help': f"停止服务"
            },
        },
    },
    'args': {
        '-d': {
            'opts': {'action': 'store_true', 'help': '显示当前目录'},
            'behavior': {'callback': current_dir, 'args': []}
        },
        '-v': {
            'opts': {'action': 'store_true', 'help': '系统版本'},
            'behavior': {'callback': version, 'args': []}
        },
    }
}
"""
命令行以及内部命令详细配置。
"""


def init_argument():
    """
    解析命令行参数，填充命令参数名字空间。
    """
    # 初始化内部命令
    for _cmd_name, _command in tool_config.get('cmds', {}).items():
        _parser = ArgSubparsers.add_parser(_cmd_name, **_command.get('opts', {}))
        for _arg_name, _argument in _command.get('args', {}).items():
            assert isinstance(_argument, dict)
            _parser.add_argument(_arg_name, **_argument)

    # 初始化选项
    for _arg_name, _argument in tool_config.get('args', {}).items():
        ArgParser.add_argument(_arg_name, **_argument.get('opts', {}))

    global ArgNamespace
    ArgNamespace = ArgParser.parse_args(sys.argv[1:])


def arg(arg_name: str):
    """
    从命令参数名字空间中获取值。

    :param arg_name: 选项名称
    :return: 选项值
    """
    global ArgNamespace
    return getattr(ArgNamespace, arg_name, None)


def exec_behavior():
    """
    执行行为方法。
    """
    sub_command_name = arg('sub_command')
    sub_command = udict.get_by_path(tool_config, f'cmds.{sub_command_name}', None)

    if sub_command_name is not None and sub_command is not None:
        # 执行命令行为
        # 回调函数
        _callback = udict.get_by_path(sub_command, 'behavior', None)
        if _callback is not None and callable(_callback):
            # 形参列表
            _arg_names = udict.get_by_path(sub_command, 'args', {}).keys()
            # 取得实参
            _kwargs = {}
            for _name in _arg_names:
                _k = _name.lstrip('-')
                _kwargs[_k] = arg(_k)
            # 验证并执行回调函数
            if len(_kwargs) == len(_arg_names):
                had_behavior(True)
                _callback(**_kwargs)
    else:
        # 执行选项行为
        for _arg_name, _argument in tool_config.get('args', {}).items():
            _name = re.sub(r'[-+/]', '', _arg_name)
            if arg(_name) is None or not arg(_name):
                # 未包含选项，跳过
                continue

            _behavior = _argument.get('behavior', None)
            if _behavior is None:
                # 未配置行为，跳出
                continue

            # 回调函数
            _callback = _behavior.get('callback', None)
            if _callback is None or not hasattr(_callback, '__call__'):
                # 行为是否是一个可用的方法或函数
                continue

            # 形参列表
            _arg_names = _behavior.get('args', [])
            # 取得实参
            _kwargs = [arg(_name) for _name in _arg_names]
            # 验证并执行回调函数
            if callable(_callback) and len(_kwargs) == len(_arg_names):
                had_behavior(True)
                _callback(*_kwargs)

    if not had_behavior():
        ArgParser.print_help()


if __name__ == '__main__':
    init_argument()
    exec_behavior()