python基于动态实例的命令处理设计实现详解

更新时间：2025年08月04日 08:27:51 作者：花酒锄作田

这篇文章主要为大家详细介绍了python基于动态实例的命令处理设计实现的相关知识,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下

前言

最近在做公司内部的一个聊天机器人服务，这个聊天机器人暂时不会用到现在热门的大模型技术，只是用于接收用户固定格式的命令，然后调用对应的方法。因为只是内部使用，所以性能也不需要太高。目前考虑的用户命令类型有以下几种：

单命令。比如用户发一个ping，调用ping主命令。
有一个子命令。比如用户发送ping version，调用ping主命令的version子命令。
单命令，带一系列位置参数。比如ping host1 host2 host3，调用ping主命令，主命令自行处理参数。
子命令有一系列位置参数。比如ping tcp host1 host2 host3，调用ping主命令的tcp子命令来处理参数。

暂不考虑子命令的子命令、flag等命令形式。

早期也没想着搞太复杂的功能，所以代码用正则表达式匹配，然后写了一堆if ... else，如今看来不是很美观，而且每次新增命令都要去配置下匹配逻辑，给别人修改时，别人经常忘了改匹配逻辑，比较繁琐。

这版的修改想法是命令类一旦声明就自动注册到某个地方，接收命令的时候自动分发到对应的命令类及其方法。想到的几个方案有监听者模式、责任链模式和本文所要提的动态实例方式（我也不知道这种方法怎么命名，瞎起了个名字）。

代码结构

│ .gitignore
│ main.py
│ README.md
│
└─commands
cmda.py
cmdb.py
__init__.py

子命令的代码都存放在./commands目录下，./commands/__init__.py声明了命令的基类，导入commands目录下除了__init__.py之外的所有python文件，以及声明工厂函数。

除了__init__.py，commands目录下的所有python文件都是命令的实现。

基类

基类的声明位于commands/__init__.py文件中，要求子类必须实现main_cmd()方法，以及通过类属性判断是否需要导入命令类。自动注册子类的方法见__init_subclass__()

from pathlib import Path
from abc import ABCMeta, abstractmethod
from threading import Lock
from collections import UserDict
import importlib
from functools import wraps
import inspect
from typing import Callable

class ThreadSafeDict(UserDict):
    """线程安全的字典"""
    def __init__(self):
        super().__init__()
        self._lock = Lock()
    
    def __setitem__(self, key, item):
        with self._lock:
            super().__setitem__(key, item)

class Command(metaclass=ABCMeta):
    registry = ThreadSafeDict()

    def __init__(self):
        # self._sub_cmds = ThreadSafeDict()
        self._sub_cmd: str = ""
        self._cmd_args: list = []

    @abstractmethod
    def main_cmd(self):
        pass

    @sub_cmd(name="help")
    def get_help(self):
        """Get help info"""
        message = f"Usage: {self._main_name} [subcommand] [args]\n"
        for name, f in self._sub_cmds.items():
            doc = f.__doc__ or ""
            message += f"  {name}, {doc}\n"
        print(message)

    def parse_cmd(self):
        cmd_list = self.command.split(" ")
        cmd_list_length = len(cmd_list)
        if cmd_list_length == 1:
            self._sub_cmd = ""
            self._cmd_args = []
        elif cmd_list_length >= 2 and cmd_list[1] not in self._sub_cmds:
            self._sub_cmd = ""
            self._cmd_args = cmd_list[1:]
        elif cmd_list_length >= 2 and cmd_list[1] in self._sub_cmds:
            self._sub_cmd = cmd_list[1]
            self._cmd_args = cmd_list[2:]
        else:
            self._sub_cmd = ""
            self._cmd_args = []

    def dispatch_command(self) -> Callable:
        """
        根据主命令和子命令的名称分发到相应的命令处理方法

        Returns:
            Callable: 返回对应的命令处理方法, 如果找不到匹配的子命令则返回 None
        """
        if not self._sub_cmd and not self._cmd_args:
            return self.main_cmd
        elif not self._sub_cmd and self._cmd_args:
            return self.main_cmd
        elif self._sub_cmd and self._sub_cmd not in self._sub_cmds:
            return None
        else:
            return self._sub_cmds[self._sub_cmd]
        
    def run(self):
        self.parse_cmd()
        func = self.dispatch_command()
        if not func:
            self.get_help()
        else:
            func(self)

	def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls_main_name = getattr(cls, "_main_name", "")
        cls_enabled = getattr(cls, "_enabled", False)
        cls_description = getattr(cls, "_description", "")
        if cls_main_name and cls_enabled and cls_description:
            cls.registry[cls._main_name.lower()] = cls  # 自动注册子类
            if not hasattr(cls, "_sub_cmds"):
                cls._sub_cmds = ThreadSafeDict()
            for name, method in inspect.getmembers(cls, inspect.isfunction):
                if hasattr(method, "__sub_cmd__"):
                    cls._sub_cmds[method.__sub_cmd__] = method
        else:
            print(f"{cls.__name__} 未注册，请检查类属性 _main_name, _enabled, _description")

子类只有导入时才会自动注册，所以写了个遍历目录进行导入的函数。

def load_commands(dir_path: Path) -> None:
    """遍历目录下的所有python文件并导入"""
    commands_dir = Path(dir_path)
    for py_file in commands_dir.glob("*.py"):
        if py_file.stem in ("__init__"):
            continue
        module_name = f"commands.{py_file.stem}"
        try:
            importlib.import_module(module_name)
        except ImportError as e:
            print(f"Failed to import {module_name}: {e}")

load_commands(Path(__file__).parent)

子命令装饰器

命令类可以使用装饰器来注册子命令，其实只是给函数加个属性。

def sub_cmd(name: str):
    """
    装饰器函数, 用于包装目标函数并添加 __sub_cmd 属性

    Args:
        name (str): 子命令名称
    """
    def decorator(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            return func(self, *args, **kwargs)
        wrapper.__sub_cmd__ = name
        return wrapper
    return decorator

实现命令类

随便写两个命令类。命令类必须声明_main_name、_enabled和_description这三个类属性，否则不会注册这个命令类。

cmda

代码文件为commands/cmda.py

from commands import Command, sub_cmd

class Cmda(Command):
    _main_name = "cmda"
    _enabled = True
    _description = "this is cmda"

    def __init__(self, command: str):
        self.command = command
        super().__init__()

    def main_cmd(self, *args: tuple, **kwargs):
        print("this is main cmd for cmda")
    
    @sub_cmd(name="info")
    def get_info(self):
        """Get info"""
        print(f"this is cmda's info")

cmdb

代码文件为commands/cmdb.py

from commands import Command, sub_cmd

class Cmdb(Command):
    _main_name = "cmdb"
    _enabled = True
    _description = "this is cmdb"

    def __init__(self, command: str):
        self.command = command
        super().__init__()

    def main_cmd(self, *args, **kwargs):
        print("this is cmdb main")


    @sub_cmd("info")
    def get_info(self):
        print("this is cmdb info")
        if self._cmd_args:
            print(f"args: {self._cmd_args}")

工厂函数

工厂函数的代码也是位于commands/__init__.py

def create_command(command: str) -> Command:
    """工厂函数"""
    if not command:
        raise ValueError("command can not be empty")
    command_list = command.split(" ")
    command_type = command_list[0]
    cls = Command.registry.get(command_type.lower())
    if not cls:
        raise ValueError(f"Unknown command: {command_type}")
    return cls(command)

使用示例

使用示例的代码位于main.py

from commands import create_command

if __name__ == '__main__':
    command = create_command("cmdb info aaa")
    command.run()
    command = create_command("cmda help")
    command.run()

执行输出

this is cmdb info
args: ['aaa']
Usage: cmda [subcommand] [args]
help, Get help info
info, Get info

完整代码

除了commands/__init__.py，其它代码文件的完整内容上面都有了，所以补充下__init__.py的内容

from pathlib import Path
from abc import ABCMeta, abstractmethod
from threading import Lock
from collections import UserDict
import importlib
from functools import wraps
import inspect
from typing import Callable

def sub_cmd(name: str):
    """
    装饰器函数, 用于包装目标函数并添加 __sub_cmd 属性

    Args:
        name (str): 子命令名称
    """
    def decorator(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            return func(self, *args, **kwargs)
        wrapper.__sub_cmd__ = name
        return wrapper
    return decorator

class ThreadSafeDict(UserDict):
    """线程安全的字典"""
    def __init__(self):
        super().__init__()
        self._lock = Lock()
    
    def __setitem__(self, key, item):
        with self._lock:
            super().__setitem__(key, item)

class Command(metaclass=ABCMeta):
    registry = ThreadSafeDict()

    def __init__(self):
        # self._sub_cmds = ThreadSafeDict()
        self._sub_cmd: str = ""
        self._cmd_args: list = []

    @abstractmethod
    def main_cmd(self):
        pass

    @sub_cmd(name="help")
    def get_help(self):
        """Get help info"""
        message = f"Usage: {self._main_name} [subcommand] [args]\n"
        for name, f in self._sub_cmds.items():
            doc = f.__doc__ or ""
            message += f"  {name}, {doc}\n"
        print(message)

    def parse_cmd(self):
        cmd_list = self.command.split(" ")
        cmd_list_length = len(cmd_list)
        if cmd_list_length == 1:
            self._sub_cmd = ""
            self._cmd_args = []
        elif cmd_list_length >= 2 and cmd_list[1] not in self._sub_cmds:
            self._sub_cmd = ""
            self._cmd_args = cmd_list[1:]
        elif cmd_list_length >= 2 and cmd_list[1] in self._sub_cmds:
            self._sub_cmd = cmd_list[1]
            self._cmd_args = cmd_list[2:]
        else:
            self._sub_cmd = ""
            self._cmd_args = []

    def dispatch_command(self) -> Callable:
        """
        根据主命令和子命令的名称分发到相应的命令处理方法

        Returns:
            Callable: 返回对应的命令处理方法, 如果找不到匹配的子命令则返回 None
        """
        if not self._sub_cmd and not self._cmd_args:
            return self.main_cmd
        elif not self._sub_cmd and self._cmd_args:
            return self.main_cmd
        elif self._sub_cmd and self._sub_cmd not in self._sub_cmds:
            return None
        else:
            return self._sub_cmds[self._sub_cmd]
        
    def run(self):
        self.parse_cmd()
        func = self.dispatch_command()
        if not func:
            self.get_help()
        else:
            func(self)



    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls_main_name = getattr(cls, "_main_name", "")
        cls_enabled = getattr(cls, "_enabled", False)
        cls_description = getattr(cls, "_description", "")
        if cls_main_name and cls_enabled and cls_description:
            cls.registry[cls._main_name.lower()] = cls  # 自动注册子类
            if not hasattr(cls, "_sub_cmds"):
                cls._sub_cmds = ThreadSafeDict()
            for name, method in inspect.getmembers(cls, inspect.isfunction):
                if hasattr(method, "__sub_cmd__"):
                    cls._sub_cmds[method.__sub_cmd__] = method
        else:
            print(f"{cls.__name__} 未注册，请检查类属性 _main_name, _enabled, _description")

def create_command(command: str) -> Command:
    """工厂函数"""
    if not command:
        raise ValueError("command can not be empty")
    command_list = command.split(" ")
    command_type = command_list[0]
    cls = Command.registry.get(command_type.lower())
    if not cls:
        raise ValueError(f"Unknown command: {command_type}")
    return cls(command)

def load_commands(dir_path: Path) -> None:
    """遍历目录下的所有python文件并导入"""
    commands_dir = Path(dir_path)
    for py_file in commands_dir.glob("*.py"):
        if py_file.stem in ("__init__"):
            continue
        module_name = f"commands.{py_file.stem}"
        try:
            importlib.import_module(module_name)
        except ImportError as e:
            print(f"Failed to import {module_name}: {e}")

load_commands(Path(__file__).parent)

__all__ = [
    "create_command",
]

到此这篇关于python基于动态实例的命令处理设计实现详解的文章就介绍到这了,更多相关python命令处理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Python eval()和exec()函数使用详解
exec函数执行的是python语句，没有返回值，eval函数执行的是python表达式，有返回值，exec函数和eval函数都可以传入命名空间作为参数，本文给大家介绍下Python eval()和exec()函数，感兴趣的朋友跟随小编一起看看吧
2022-11-11
Python接入MySQL实现增删改查的实战记录
这篇文章主要给大家介绍了关于Python接入MySQL实现增删改查的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2021-03-03
Python Barbershop实现照片换发型功能
这篇文章主要为大家介绍了一个开源项目（Barbershop），可以将照片中的发型更换成另一个，文中实现过程讲解详细，感兴趣的可以学习一下
2022-01-01
如何用python获取到照片拍摄时的详细位置(附源码)
其实我们平时拍摄的照片里,隐藏了大量的信息,下面这篇文章主要给大家介绍了关于如何用python获取到照片拍摄时的详细位置,文中通过实例代码介绍的非常详细,需要的朋友可以参考下
2022-12-12
关于数据分析之滚动窗口pandas.DataFrame.rolling方法
Pandas库中的rolling方法是数据处理中常用的功能,它允许用户对数据进行滚动窗口（滑动窗口）操作,通过指定窗口大小,可以使用不同的聚合函数对窗口内的数据进行计算,例如最大值、最小值、平均值、中位数等,此外,rolling方法还可以计算方差、标准差、偏度、峰度
2024-09-09
django template实现定义临时变量,自定义赋值、自增实例
这篇文章主要介绍了django template实现定义临时变量,自定义赋值、自增实例，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-07-07
python如何调用外部的exe程序
本文介绍了在Python中执行外部exe命令时遇到的问题及解决方法,包括路径写法、中文输出乱码以及文件编码等问题,并提供了一些个人经验
2025-02-02
window下eclipse安装python插件教程
这篇文章主要为大家详细介绍了window下eclipse安装python插件教程，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2017-04-04
Python基础之numpy库的使用
这篇文章主要介绍了Python基础之numpy库的使用,文中有非常详细的代码示例,对正在学习python基础的小伙伴们有非常好的帮助,需要的朋友可以参考下
2021-04-04
Python urllib.request对象案例解析
这篇文章主要介绍了Python urllib.request对象案例解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-05-05