python基于動(dòng)態(tài)實(shí)例的命令處理設(shè)計(jì)實(shí)現(xiàn)詳解

更新時(shí)間：2025年08月04日 08:27:51 作者：花酒鋤作田

這篇文章主要為大家詳細(xì)介紹了python基于動(dòng)態(tài)實(shí)例的命令處理設(shè)計(jì)實(shí)現(xiàn)的相關(guān)知識(shí),文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

前言

最近在做公司內(nèi)部的一個(gè)聊天機(jī)器人服務(wù)，這個(gè)聊天機(jī)器人暫時(shí)不會(huì)用到現(xiàn)在熱門的大模型技術(shù)，只是用于接收用戶固定格式的命令，然后調(diào)用對(duì)應(yīng)的方法。因?yàn)橹皇莾?nèi)部使用，所以性能也不需要太高。目前考慮的用戶命令類型有以下幾種：

單命令。比如用戶發(fā)一個(gè)ping，調(diào)用ping主命令。
有一個(gè)子命令。比如用戶發(fā)送ping version，調(diào)用ping主命令的version子命令。
單命令，帶一系列位置參數(shù)。比如ping host1 host2 host3，調(diào)用ping主命令，主命令自行處理參數(shù)。
子命令有一系列位置參數(shù)。比如ping tcp host1 host2 host3，調(diào)用ping主命令的tcp子命令來處理參數(shù)。

暫不考慮子命令的子命令、flag等命令形式。

早期也沒想著搞太復(fù)雜的功能，所以代碼用正則表達(dá)式匹配，然后寫了一堆if ... else，如今看來不是很美觀，而且每次新增命令都要去配置下匹配邏輯，給別人修改時(shí)，別人經(jīng)常忘了改匹配邏輯，比較繁瑣。

這版的修改想法是命令類一旦聲明就自動(dòng)注冊(cè)到某個(gè)地方，接收命令的時(shí)候自動(dòng)分發(fā)到對(duì)應(yīng)的命令類及其方法。想到的幾個(gè)方案有監(jiān)聽者模式、責(zé)任鏈模式和本文所要提的動(dòng)態(tài)實(shí)例方式（我也不知道這種方法怎么命名，瞎起了個(gè)名字）。

代碼結(jié)構(gòu)

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

子命令的代碼都存放在./commands目錄下，./commands/__init__.py聲明了命令的基類，導(dǎo)入commands目錄下除了__init__.py之外的所有python文件，以及聲明工廠函數(shù)。

除了__init__.py，commands目錄下的所有python文件都是命令的實(shí)現(xiàn)。

基類

基類的聲明位于commands/__init__.py文件中，要求子類必須實(shí)現(xiàn)main_cmd()方法，以及通過類屬性判斷是否需要導(dǎo)入命令類。自動(dòng)注冊(cè)子類的方法見__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:
        """
        根據(jù)主命令和子命令的名稱分發(fā)到相應(yīng)的命令處理方法

        Returns:
            Callable: 返回對(duì)應(yīng)的命令處理方法, 如果找不到匹配的子命令則返回 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  # 自動(dòng)注冊(cè)子類
            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__} 未注冊(cè)，請(qǐng)檢查類屬性 _main_name, _enabled, _description")

子類只有導(dǎo)入時(shí)才會(huì)自動(dòng)注冊(cè)，所以寫了個(gè)遍歷目錄進(jìn)行導(dǎo)入的函數(shù)。

def load_commands(dir_path: Path) -> None:
    """遍歷目錄下的所有python文件并導(dǎo)入"""
    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)

子命令裝飾器

命令類可以使用裝飾器來注冊(cè)子命令，其實(shí)只是給函數(shù)加個(gè)屬性。

def sub_cmd(name: str):
    """
    裝飾器函數(shù), 用于包裝目標(biāo)函數(shù)并添加 __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

實(shí)現(xiàn)命令類

隨便寫兩個(gè)命令類。命令類必須聲明_main_name、_enabled和_description這三個(gè)類屬性，否則不會(huì)注冊(cè)這個(gè)命令類。

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}")

工廠函數(shù)

工廠函數(shù)的代碼也是位于commands/__init__.py

def create_command(command: str) -> Command:
    """工廠函數(shù)"""
    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()

執(zhí)行輸出

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

完整代碼

除了commands/__init__.py，其它代碼文件的完整內(nèi)容上面都有了，所以補(bǔ)充下__init__.py的內(nèi)容

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):
    """
    裝飾器函數(shù), 用于包裝目標(biāo)函數(shù)并添加 __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:
        """
        根據(jù)主命令和子命令的名稱分發(fā)到相應(yīng)的命令處理方法

        Returns:
            Callable: 返回對(duì)應(yīng)的命令處理方法, 如果找不到匹配的子命令則返回 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  # 自動(dòng)注冊(cè)子類
            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__} 未注冊(cè)，請(qǐng)檢查類屬性 _main_name, _enabled, _description")

def create_command(command: str) -> Command:
    """工廠函數(shù)"""
    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文件并導(dǎo)入"""
    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",
]

到此這篇關(guān)于python基于動(dòng)態(tài)實(shí)例的命令處理設(shè)計(jì)實(shí)現(xiàn)詳解的文章就介紹到這了,更多相關(guān)python命令處理內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: