import os import importlib.util import sys # import custom_plugins class LoadAllPlugins(object): def __init__(self, windows): self.windows = windows self.ignore_files = ["get_remote_has256.py", "__init__.py"] self.ignore_dirs = ["other_micropython_code", "__pycache__"] def send_log(self, text): if self.windows: self.windows.send_log(text) def load_plugins(self, plugin_dir): """ 递归加载插件目录下的所有包和模块。 """ plugin_modules = {} # 获取插件目录的绝对路径 plugin_dir_abs = os.path.abspath(plugin_dir) print("22 plugin_dir_abs", plugin_dir_abs) # # 如果插件目录不在sys.path中,则添加它 if plugin_dir_abs not in sys.path: sys.path.append(plugin_dir_abs) def load_module_from_file(module_name, file_path): try: spec = importlib.util.spec_from_file_location(module_name, file_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) print(f"Loaded module Successful: {module_name}") self.send_log(f"Loaded module Successful: {module_name}") return module except Exception as e: print(f"Failed to load module {module_name}: {e}") self.send_log(f"Failed to load module {module_name}: {e}") return None def traverse_directory(directory): for root, dirs, files in os.walk(directory): f = True for i in self.ignore_dirs: if i in root: f = False break if f is False: continue # 确保当前目录被视为Python包 init_file = os.path.join(root, '__init__.py') if not os.path.isfile(init_file): open(init_file, 'a').close() # 创建空的 __init__.py 文件 # 加载当前目录下的所有 .py 文件(不包括 __init__.py) for file in files: if file in self.ignore_files: continue if file.endswith('.py'): module_name = \ os.path.relpath(os.path.join(root, file), plugin_dir).replace(os.sep, '.').rsplit('.', 1)[0] module = load_module_from_file(module_name, os.path.join(root, file)) if module is not None: plugin_modules[module_name] = module # 遍历插件目录,寻找所有的包和模块 traverse_directory(plugin_dir_abs) return plugin_modules def do_load(self, root): plugin_b_dir = r"{}/custom_plugins".format(os.getcwd()) if plugin_b_dir not in sys.path: sys.path.insert(0, plugin_b_dir) plugins_dict = {"detail_template": {}, "ampy_scribe": {}, "generate_scribe": {}, } self.load_plugins(plugin_dir="{}\custom_plugins/plugins_mode".format(root)) plugins = self.load_plugins(plugin_dir="{}\custom_plugins/plugins".format(root)) for i, py_file in plugins.items(): print(i) if hasattr(py_file, 'plugins_name'): if py_file.plugins_name == "详情模板": company_name_list = py_file.company_name_list template_name = py_file.template_name # 一个模板文件可能适合多个不同的企业 if "全部" in company_name_list: company_name_list = ["全部"] for company_name in company_name_list: if company_name not in plugins_dict["detail_template"]: plugins_dict["detail_template"][company_name] = {} plugins_dict["detail_template"][company_name][template_name] = py_file.DetailPicGet if py_file.plugins_name == "AMPY插件": plugins_dict["ampy_scribe"]["AMPY插件"] = py_file.RemoteUpdate # if py_file.plugins_name == "主图生成与上传": # plugins_dict["generate_scribe"]["主图生成与上传"] = py_file.AutoDealPics return plugins_dict def load(self): print("Loading plugins...") plugins_dict = {} try: plugins_dict = self.do_load(root=os.getcwd()) except BaseException as e: print(e) self.send_log("Loading plugins. error {}".format(e)) # print(plugins_dict) # _class = plugins_dict["detail_template"]["小苏"]["xiaosushuoxie-1"] # _class() return plugins_dict if __name__ == '__main__': LoadAllPlugins(None).load()