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

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

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

前言

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

單命令。比如用戶發(fā)一個ping，調(diào)用ping主命令。
有一個子命令。比如用戶發(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等命令形式。

早期也沒想著搞太復雜的功能，所以代碼用正則表達式匹配，然后寫了一堆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，其它代碼文件的完整內(nèi)容上面都有了，所以補充下__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ù), 用于包裝目標函數(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命令處理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

Python?eval()和exec()函數(shù)使用詳解
exec函數(shù)執(zhí)行的是python語句，沒有返回值，eval函數(shù)執(zhí)行的是python表達式，有返回值，exec函數(shù)和eval函數(shù)都可以傳入命名空間作為參數(shù)，本文給大家介紹下Python?eval()和exec()函數(shù)，感興趣的朋友跟隨小編一起看看吧
2022-11-11
Python接入MySQL實現(xiàn)增刪改查的實戰(zhàn)記錄
這篇文章主要給大家介紹了關于Python接入MySQL實現(xiàn)增刪改查的相關資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2021-03-03
Python Barbershop實現(xiàn)照片換發(fā)型功能
這篇文章主要為大家介紹了一個開源項目（Barbershop），可以將照片中的發(fā)型更換成另一個，文中實現(xiàn)過程講解詳細，感興趣的可以學習一下
2022-01-01
如何用python獲取到照片拍攝時的詳細位置(附源碼)
其實我們平時拍攝的照片里,隱藏了大量的信息,下面這篇文章主要給大家介紹了關于如何用python獲取到照片拍攝時的詳細位置,文中通過實例代碼介紹的非常詳細,需要的朋友可以參考下
2022-12-12
關于數(shù)據(jù)分析之滾動窗口pandas.DataFrame.rolling方法
Pandas庫中的rolling方法是數(shù)據(jù)處理中常用的功能,它允許用戶對數(shù)據(jù)進行滾動窗口（滑動窗口）操作,通過指定窗口大小,可以使用不同的聚合函數(shù)對窗口內(nèi)的數(shù)據(jù)進行計算,例如最大值、最小值、平均值、中位數(shù)等,此外,rolling方法還可以計算方差、標準差、偏度、峰度
2024-09-09
django template實現(xiàn)定義臨時變量,自定義賦值、自增實例
這篇文章主要介紹了django template實現(xiàn)定義臨時變量,自定義賦值、自增實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2020-07-07
python如何調(diào)用外部的exe程序
本文介紹了在Python中執(zhí)行外部exe命令時遇到的問題及解決方法,包括路徑寫法、中文輸出亂碼以及文件編碼等問題,并提供了一些個人經(jīng)驗
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