coordinate/vs_plg_factory.py

# vs_plg_factory.py
import importlib.util
import sys
import os
from vs_plg import IVSPlg # type: ignore
from vs_log_macros import VSC_ERROR, VSC_INFO # type: ignore

class VSPlgFactory:
    """
    @brief 签证插件工厂类
    负责动态加载和创建 IVSPlg 接口的实现实例。
    """
    def __init__(self):
        self._registered_plugins = {} # {plugin_name: (module_name, class_name)}

    def register_plugin(self, plugin_name: str, module_path: str, class_name: str):
        """
        @brief 注册插件
        @param plugin_name 插件的逻辑名称
        @param module_path 插件模块的 Python 文件路径 (e.g., "plugins/concrete_plugin.py")
        @param class_name 插件实现类的名称 (e.g., "ConcretePlugin")
        """
        self._registered_plugins[plugin_name] = (module_path, class_name)
        VSC_INFO("factory", "Registered plugin '%s' from '%s' class '%s'", plugin_name, module_path, class_name)

    def create(self, group_id: str, plugin_name: str) -> IVSPlg:
        """
        @brief 创建一个 IVSPlg 实例
        @param group_id 实例所属的分组 ID
        @param plugin_name 要创建的插件的注册名称
        @return IVSPlg 实例
        """
        if plugin_name not in self._registered_plugins:
            VSC_ERROR("factory", "Plugin '%s' not registered.", plugin_name)
            raise ValueError(f"Plugin '{plugin_name}' not registered.")
        
        module_path, class_name = self._registered_plugins[plugin_name]
        
        try:
            # 动态加载模块
            spec = importlib.util.spec_from_file_location(plugin_name, module_path)
            if spec is None:
                VSC_ERROR("factory", "Failed to find module spec for '%s' at '%s'", plugin_name, module_path)
                raise ImportError(f"Could not find module spec for {plugin_name} at {module_path}")
            
            module = importlib.util.module_from_spec(spec)
            sys.modules[plugin_name] = module
            spec.loader.exec_module(module)
            
            # 获取类并实例化
            plugin_class = getattr(module, class_name)
            instance = plugin_class(group_id)
            
            if not isinstance(instance, IVSPlg):
                VSC_ERROR("factory", "Created instance for '%s' is not an IVSPlg.", plugin_name)
                raise TypeError(f"Plugin '{plugin_name}' class '{class_name}' does not implement IVSPlg.")
            
            VSC_INFO("factory", "Successfully created instance for group '%s' with plugin '%s'", group_id, plugin_name)
            return instance

        except Exception as e:
            VSC_ERROR("factory", "Failed to create plugin instance '%s': %s", plugin_name, str(e))
            raise