127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
import os
|
|
import sys
|
|
import time
|
|
import logging
|
|
import importlib
|
|
import threading
|
|
|
|
|
|
from pathlib import Path
|
|
from enum import Enum, auto
|
|
from PySide6.QtCore import QObject,Signal
|
|
from dataclasses import dataclass, field, fields
|
|
|
|
LOCAL_PLUGINS_PATH = Path("./plugins")
|
|
REMOTE_PLUGINS_PATH = Path('Y:/SE/xufeifei/plugins')
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Event(Enum):
|
|
Update = auto()
|
|
Install = auto()
|
|
Uninstall = auto()
|
|
|
|
class PluginRegistry(QObject):
|
|
"""插件注册中心。"""
|
|
plugins_loader_signal = Signal(dict)
|
|
update_plugins_card_signal = Signal(dict)
|
|
def __init__(self, parent = None) -> None:
|
|
super().__init__(parent)
|
|
|
|
self.plugins = dict()
|
|
|
|
self.start_discover_local(LOCAL_PLUGINS_PATH, self.plugins)
|
|
self.start_discover_remote(REMOTE_PLUGINS_PATH, self.plugins)
|
|
|
|
def start_discover_local(self,plugins_dir: Path, plugins:dict):
|
|
self.thread_local = threading.Thread(target= self.discover_local_event,args=(plugins_dir, plugins,))
|
|
self.thread_local.start()
|
|
|
|
def discover_local_event(self, plugins_dir: Path, plugins:dict):
|
|
if not os.path.exists(plugins_dir):
|
|
os.mkdir(plugins_dir)
|
|
return
|
|
|
|
sys.path.append(str(plugins_dir))
|
|
for tool in Path(plugins_dir).glob("*.pyd"):
|
|
tool_name = tool.stem
|
|
try:
|
|
module = importlib.import_module(tool_name)
|
|
obj = module.create_plugin()
|
|
name = module.read_plugin_name()
|
|
version = module.read_plugin_version()
|
|
if name in plugins.keys():
|
|
pass
|
|
else:
|
|
plugins[name] = {}
|
|
plugins[name]['obj'] = obj
|
|
plugins[name]['local version'] = version
|
|
except Exception as e:
|
|
logger.error(f"插件{tool_name}注册失败，{str(e)}")
|
|
sys.path.remove(str(LOCAL_PLUGINS_PATH))
|
|
self.plugins_loader_signal.emit(plugins)
|
|
|
|
|
|
def start_discover_remote(self,plugins_dir: Path, plugins:dict):
|
|
self.thread_remote = threading.Thread(target= self.discover_remote_event,args=(plugins_dir, plugins,))
|
|
self.thread_remote.start()
|
|
|
|
def discover_remote_event(self, plugins_dir: Path, plugins:dict):
|
|
if not os.path.exists(plugins_dir):
|
|
logger.error("服务器链接错误！")
|
|
return
|
|
#读取服务器工具信息
|
|
time.sleep(0.01)
|
|
sys.path.append(str(plugins_dir))
|
|
for tool in Path(plugins_dir).glob("*.pyd"):
|
|
tool_name = tool.stem
|
|
try:
|
|
module = importlib.import_module(tool_name)
|
|
|
|
name = module.read_plugin_name()
|
|
version = module.read_plugin_version()
|
|
description = module.read_plugin_description()
|
|
if name in plugins.keys():
|
|
pass
|
|
else:
|
|
print("discover_remote_event")
|
|
plugins[name] = {}
|
|
plugins[name]["tool_name"] = tool_name
|
|
plugins[name]["local version"] = None
|
|
plugins[name]["local description"] = None
|
|
plugins[name]["remote version"] = version
|
|
plugins[name]["remote description"] = description
|
|
plugins[name]['remote path'] = plugins_dir
|
|
except Exception as e:
|
|
logger.error(f"服务器{tool_name}无法加载：{str(e)}")
|
|
sys.path.remove(str(plugins_dir))
|
|
self.update_plugins_card_signal.emit(plugins)
|
|
|
|
def start_plugins_event(self, event:Event, name:str):
|
|
if event == Event.Install:
|
|
local_path = str(LOCAL_PLUGINS_PATH) + '\\' + f"{self.plugins[name]['tool_name']}.pyd"
|
|
remote_path = str(REMOTE_PLUGINS_PATH) + '\\' + f"{self.plugins[name]['tool_name']}.pyd"
|
|
|
|
self.sub_thread = threading.Thread(target= self.install_plugins_event,args=(local_path,remote_path,))
|
|
self.sub_thread.start()
|
|
pass
|
|
|
|
def install_plugins_event(self, local_path, remote_path):
|
|
try:
|
|
total_size = os.path.getsize(remote_path)
|
|
copied_size = 0
|
|
|
|
with open(remote_path, 'rb') as fsrc, open(local_path, 'wb') as fdst:
|
|
while True:
|
|
buf = fsrc.read(1024*1024)
|
|
if not buf:
|
|
break
|
|
fdst.write(buf)
|
|
copied_size += len(buf)
|
|
logger.info(f"安装进度: {copied_size/1024/1024:.2f} MB / {total_size/1024/1024:.2f} MB")
|
|
time.sleep(0.1)
|
|
logger.info(f"{local_path}安装成功!!!")
|
|
except Exception as e:
|
|
logger.error(f"{remote_path}安装失败{e.args}")
|
|
return |