开源小项目：pyouter 0.0.1 发布

Posted 2022-12-16 幻灰龙

发布一个业余开源小项目 pyouter 。

• python Github 仓库：https://github.com/fanfeilong/task_router
• pip 包安装：https://pypi.org/project/pyouter/0.0.1
• contributors: @幻灰龙 @ccat

传统的命令行参数一般会设计 -x, --xxx这样的命令解析。一个痛点是命令多了之后就难以记忆和使用。同时不易于通过命令行定向控制任务执行。

受 HTTP Rest API path和路由器的影响。最开始我在项目中让命令行采用基于点分隔符来完成 path 的功能，例如：

python main.py -a ocr
python main.py -a ask
python main.py -a ask.answer
python main.py -a ask.skill_tree
python main.py -a ask.skill_tree.match

实现上解析 ask.skill_tree.match 后就可以有层次地路由，在内部的路由映射表里递归查找，一直找到叶子节点就执行叶子节点的配置任务。

最开始的路由解析代码非常简单：

def dispatch(config, options, actions, targets):
    ''' 分发命令行 action '''
    action_len = len(actions)
    action_len = len(actions)
    if action_len < 2:
        return

    index = 1
    next = targets
    action = actions[index]
    print(f"[命令路由中..]: actions[0]")

    while action_len >= index:
        if type(next) == type():
            if index == action_len:
                if next.get('run') != None:
                    print(f"[命令路由执行]:", '->'.join(actions))
                    next['run']()
                    break

            action = actions[index]
            if next.get(action) != None:
                print(f"[命令路由中..]: action")
                next = next[action]
                index += 1
            else:
                print("[命令路由错误]: 未找到支持的命令行路由：", '->'.join(actions))
                index += 1
        else:
            print(f"[命令路由执行]:", '->'.join(actions))

            next()
            index += 1
            break

用例：

if __name__=='__main__':
	config = ...
	options = ...
	dispatch(config, options, actions, 
	    "ask": 
	        "run": lambda: dispatch_ask(config, options),
	        "answer": lambda: dispatch_answer(config, options),
	        "code": lambda: dispatch_code(config, options),
	        "tag": lambda: dispatch_tag(config, options),
	        "title": lambda: dispatch_title(config, options),
	        "skill_tree": 
	            "run": lambda: dispatch_skill_tree(config, options),
	            "main": lambda: dispatch_skill_tree_main(config, options),
	            "match": lambda: dispatch_skill_tree_match(config, options),
	        ,
	        "book": lambda: dispatch_book(config, options),
	        "level": lambda: dispatch_level(config, options),
	        "pycorrector": lambda: dispatch_pycorrector(config, options),
	        "tag_index": lambda: dispatch_tag_index(config, options)
	    ,
	    "ocr": lambda: dispatch_ocr(config, options),
	    "bert": lambda: dispatch_bert(config, options),
	    "blog_tag_clean": lambda: dispatch_blog_tag_clean(config, options)
	)

@ccat 同学觉的这个想法很好，非常勤快地用对象的方式基于递归重新实现了一个版本路由器。

from .errors import NotFound


class Router(object):
    def __init__(self, **args):
        self.route = 
        for key in args:
            self.route[key] = args[key]

    def context(self, config, options):
        self.config = config
        self.options = options
        for key in self.route:
            router = self.route[key]
            if type(router) == type(self):
                router.context(config, options)

    def dispatch(self, command: str):
        if "." in command:
            crt, nxt = command.split('.', 1)
            if crt not in self.route:
                raise NotFound(self.route, crt)
            if self.options.view:
                print(f'->router: crt')
            return self.route[crt].dispatch(nxt)
        else:
            if self.options.view:
                print(f'->action: command')
            if command not in self.route.keys():
                raise NotFound(self.route, command)

            return self.route[command](self.config, self.options)

    def tasks(self, base=None):

        for key in self.route:
            current = f"base.key" if base else key
            item = self.route[key]
            if type(item) is Router:
                for task in item.tasks(current):
                    yield task
            else:
                yield current

我们创建了一个仓库：https://github.com/fanfeilong/task_router 以后可以在CodeChina上也同步一个。这个版本的实现很简洁，于是我也贡献了App组织层：

from pyouter.default import create_parser
from pyouter.errors import NotInit
from pyouter.router import Router


class App(object):
    def __init__(self, **args):
        opt_parser = create_parser("tasks router runner")
        self.options = opt_parser.parse_args()
        self.config = 
        self.router: Router

    def use(self, router: Router):
        self.router = router
        self.router.context(self.config, self.options)
        return self

    def run(self):
        if self.router is None:
            raise NotInit("self.router in App")

        if self.options.tasks:
            for task in self.router.tasks():
                print(task)
        else:
            self.router.dispatch(self.options.actions)
        return self

这样最终使用起来长这样：

from pyouter.router import Router
from pyouter.oh_my_zsh.install import install as omz
from pyouter.fish.install import install as fish
from app import App


if __name__ == "__main__":
    app = App()
    app.use(
        router=Router(
            install=Router(
                ohmyzsh=omz,
                fish=fish)))
    app.run()

用例：

python main.py router.install.fish
python main.py router.install.ohmyzsh

pip 类库安装：pip install pyouter

第1版本比较简单，但是核心设计思想已经定了，更多丰富的功能会持续更新。欢迎试用！不要让 Python 代码变成一堆碎片脚本，请用pyouter组织你的Python脚本任务，提供一个有层次的易于记忆的命令行参数。