zhaoyuan_DataPreProcess/core/plugin_manager.py

import importlib
import inspect
from pathlib import Path
from typing import Dict, List, Type, Optional
import sys

from .plugin_interface import CustomDataProcessorPlugin

class PluginManager:
    """Manages the discovery and loading of data processor plugins."""
    
    def __init__(self, plugin_dir: Path):
        self.plugin_dir = plugin_dir
        self.plugins: Dict[str, Type[CustomDataProcessorPlugin]] = {}
        self._load_plugins()
    
    def _load_plugins(self) -> None:
        """发现并加载插件目录中的所有插件"""
        if not self.plugin_dir.exists():
            print(f"未找到插件目录: {self.plugin_dir}")
            return
            
        # 将插件目录添加到Python路径
        sys.path.insert(0, str(self.plugin_dir.parent))
        
        try:
            # 查找插件目录中的Python文件
            for plugin_file in self.plugin_dir.glob("*.py"):
                if plugin_file.name == "__init__.py":
                    continue
                    
                try:
                    # 修改模块导入方式，直接使用文件名
                    module_name = f"plugins.{plugin_file.stem}"
                    print(f"尝试加载插件模块: {module_name}")
                        
                    module = importlib.import_module(module_name)
                    
                    # 在模块中查找插件类
                    for name, obj in inspect.getmembers(module):
                        if (inspect.isclass(obj) 
                            and issubclass(obj, CustomDataProcessorPlugin) 
                            and obj != CustomDataProcessorPlugin):
                            # 从类名中获取关键词（移除DataProcessor或Processor后缀）
                            keyword = name.lower().replace('dataprocessor', '').replace('processor', '')
                            self.plugins[keyword] = obj
                            print(f"成功加载插件: {name}, 关键词: {keyword}")
                            
                except ImportError as e:
                    print(f"导入插件 {plugin_file.name} 失败: {e}")
                except Exception as e:
                    print(f"加载插件 {plugin_file.name} 时出错: {e}")
                    
        finally:
            # 从Python路径中移除插件目录
            if str(self.plugin_dir.parent) in sys.path:
                sys.path.remove(str(self.plugin_dir.parent))
            
    def get_plugins(self) -> List[Type[CustomDataProcessorPlugin]]:
        """Returns a list of all loaded plugin classes."""
        return list(self.plugins.values())
    
    def find_plugin_by_folder_name(self, folder_name: str) -> Optional[Type[CustomDataProcessorPlugin]]:
        """根据文件夹名称查找匹配的插件"""
        folder_name = folder_name.lower()
        for keyword, plugin_class in self.plugins.items():
            if keyword in folder_name:
                return plugin_class
        return None
    
    def get_plugin_for_data(self, zip_path: Path, folder_name: str) -> Type[CustomDataProcessorPlugin]:
        """查找能处理给定数据的插件"""
        # 首先通过文件夹名称匹配插件
        plugin_class = self.find_plugin_by_folder_name(folder_name)
        if plugin_class:
            try:
                plugin = plugin_class()
                if plugin.can_handle(zip_path, folder_name):
                    return plugin_class
            except Exception as e:
                print(f"检查插件 {plugin_class.__name__} 时出错: {e}")
                
        return None