python基于動態(tài)實例的命令處理設計實現(xiàn)詳解

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

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

前言

最近在做公司內部的一個聊天機器人服務，這個聊天機器人暫時不會用到現(xiàn)在熱門的大模型技術，只是用于接收用戶固定格式的命令，然后調用對應的方法。因為只是內部使用，所以性能也不需要太高。目前考慮的用戶命令類型有以下幾種：

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

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

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

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

代碼結構

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

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

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

基類

基類的聲明位于commands/__init__.py文件中，要求子類必須實現(xiàn)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:
        """
        根據(jù)主命令和子命令的名稱分發(fā)到相應的命令處理方法

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

子類只有導入時才會自動注冊，所以寫了個遍歷目錄進行導入的函數(shù)。

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)

子命令裝飾器

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

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

實現(xiàn)命令類

隨便寫兩個命令類。命令類必須聲明_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}")

工廠函數(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，其它代碼文件的完整內容上面都有了，所以補充下__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):
    """
    裝飾器函數(shù), 用于包裝目標函數(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ā)到相應的命令處理方法

        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:
    """工廠函數(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文件并導入"""
    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基于動態(tài)實例的命令處理設計實現(xiàn)詳解的文章就介紹到這了,更多相關python命令處理內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

python利器pyproject.toml詳解
pyproject.toml是Python項目管理的??核心樞紐??,通過??統(tǒng)一配置??解決傳統(tǒng)方案的碎片化問題,提升構建可靠性、依賴管理靈活性和工具鏈集成度,本文給大家介紹python利器pyproject.toml的相關知識,感興趣的朋友跟隨小編一起看看吧
2025-09-09
python可視化實現(xiàn)KNN算法
這篇文章主要為大家詳細介紹了python可視化實現(xiàn)KNN算法，通過繪圖工具Matplotlib包可視化實現(xiàn)機器學習中的KNN算法，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2019-10-10
python數(shù)組復制拷貝的實現(xiàn)方法
這篇文章主要介紹了python數(shù)組復制拷貝的實現(xiàn)方法,實例分析了Python數(shù)組傳地址與傳值兩種復制拷貝的使用技巧,需要的朋友可以參考下
2015-06-06
最基礎的Python的socket編程入門教程
這篇文章主要介紹了最基礎的Python的socket編程入門教程,包括最基本的發(fā)送和接受信息等內容,需要的朋友可以參考下
2015-04-04
Python實現(xiàn)特定場景去除高光算法詳解
這篇文章主要介紹了如何利用Python+OpenCV實現(xiàn)特定場景去除高光算法，文中的示例代碼講解詳細，對我們學習Python有一定的幫助，需要的可以參考一下
2021-12-12
Pandas之數(shù)據(jù)追加df.append方式
這篇文章主要介紹了Pandas之數(shù)據(jù)追加df.append方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教
2023-08-08
Python queue隊列原理與應用案例分析
這篇文章主要介紹了Python queue隊列原理與應用,結合具體案例形式分析了Python queue隊列的原理、功能、實現(xiàn)方法與使用技巧,需要的朋友可以參考下
2019-09-09
python3實現(xiàn)小球轉動抽獎小游戲
這篇文章主要為大家詳細介紹了python3實現(xiàn)小球轉動抽獎小游戲，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2019-05-05
python隊列queue模塊詳解
這篇文章主要為大家詳細介紹了python隊列queue模塊的相關資料，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2018-04-04
淺談django 重載str 方法
這篇文章主要介紹了淺談django 重載str 方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2020-05-05