File size: 14,455 Bytes
61517de |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 |
# encoding:utf-8
import importlib
import importlib.util
import json
import os
import sys
from common.log import logger
from common.singleton import singleton
from common.sorted_dict import SortedDict
from config import conf, write_plugin_config
from .event import *
@singleton
class PluginManager:
def __init__(self):
self.plugins = SortedDict(lambda k, v: v.priority, reverse=True)
self.listening_plugins = {}
self.instances = {}
self.pconf = {}
self.current_plugin_path = None
self.loaded = {}
def register(self, name: str, desire_priority: int = 0, **kwargs):
def wrapper(plugincls):
plugincls.name = name
plugincls.priority = desire_priority
plugincls.desc = kwargs.get("desc")
plugincls.author = kwargs.get("author")
plugincls.path = self.current_plugin_path
plugincls.version = kwargs.get("version") if kwargs.get("version") != None else "1.0"
plugincls.namecn = kwargs.get("namecn") if kwargs.get("namecn") != None else name
plugincls.hidden = kwargs.get("hidden") if kwargs.get("hidden") != None else False
plugincls.enabled = True
if self.current_plugin_path == None:
raise Exception("Plugin path not set")
self.plugins[name.upper()] = plugincls
logger.info("Plugin %s_v%s registered, path=%s" % (name, plugincls.version, plugincls.path))
return wrapper
def save_config(self):
with open("./plugins/plugins.json", "w", encoding="utf-8") as f:
json.dump(self.pconf, f, indent=4, ensure_ascii=False)
def load_config(self):
logger.info("Loading plugins config...")
modified = False
if os.path.exists("./plugins/plugins.json"):
with open("./plugins/plugins.json", "r", encoding="utf-8") as f:
pconf = json.load(f)
pconf["plugins"] = SortedDict(lambda k, v: v["priority"], pconf["plugins"], reverse=True)
else:
modified = True
pconf = {"plugins": SortedDict(lambda k, v: v["priority"], reverse=True)}
self.pconf = pconf
if modified:
self.save_config()
return pconf
@staticmethod
def _load_all_config():
"""
背景: 目前插件配置存放于每个插件目录的config.json下,docker运行时不方便进行映射,故增加统一管理的入口,优先
加载 plugins/config.json,原插件目录下的config.json 不受影响
从 plugins/config.json 中加载所有插件的配置并写入 config.py 的全局配置中,供插件中使用
插件实例中通过 config.pconf(plugin_name) 即可获取该插件的配置
"""
all_config_path = "./plugins/config.json"
try:
if os.path.exists(all_config_path):
# read from all plugins config
with open(all_config_path, "r", encoding="utf-8") as f:
all_conf = json.load(f)
logger.info(f"load all config from plugins/config.json: {all_conf}")
# write to global config
write_plugin_config(all_conf)
except Exception as e:
logger.error(e)
def scan_plugins(self):
logger.info("Scaning plugins ...")
plugins_dir = "./plugins"
raws = [self.plugins[name] for name in self.plugins]
for plugin_name in os.listdir(plugins_dir):
plugin_path = os.path.join(plugins_dir, plugin_name)
if os.path.isdir(plugin_path):
# 判断插件是否包含同名__init__.py文件
main_module_path = os.path.join(plugin_path, "__init__.py")
if os.path.isfile(main_module_path):
# 导入插件
import_path = "plugins.{}".format(plugin_name)
try:
self.current_plugin_path = plugin_path
if plugin_path in self.loaded:
if self.loaded[plugin_path] == None:
logger.info("reload module %s" % plugin_name)
self.loaded[plugin_path] = importlib.reload(sys.modules[import_path])
dependent_module_names = [name for name in sys.modules.keys() if name.startswith(import_path + ".")]
for name in dependent_module_names:
logger.info("reload module %s" % name)
importlib.reload(sys.modules[name])
else:
self.loaded[plugin_path] = importlib.import_module(import_path)
self.current_plugin_path = None
except Exception as e:
logger.warn("Failed to import plugin %s: %s" % (plugin_name, e))
continue
pconf = self.pconf
news = [self.plugins[name] for name in self.plugins]
new_plugins = list(set(news) - set(raws))
modified = False
for name, plugincls in self.plugins.items():
rawname = plugincls.name
if rawname not in pconf["plugins"]:
modified = True
logger.info("Plugin %s not found in pconfig, adding to pconfig..." % name)
pconf["plugins"][rawname] = {
"enabled": plugincls.enabled,
"priority": plugincls.priority,
}
else:
self.plugins[name].enabled = pconf["plugins"][rawname]["enabled"]
self.plugins[name].priority = pconf["plugins"][rawname]["priority"]
self.plugins._update_heap(name) # 更新下plugins中的顺序
if modified:
self.save_config()
return new_plugins
def refresh_order(self):
for event in self.listening_plugins.keys():
self.listening_plugins[event].sort(key=lambda name: self.plugins[name].priority, reverse=True)
def activate_plugins(self): # 生成新开启的插件实例
failed_plugins = []
for name, plugincls in self.plugins.items():
if plugincls.enabled:
if name not in self.instances:
try:
instance = plugincls()
except Exception as e:
logger.warn("Failed to init %s, diabled. %s" % (name, e))
self.disable_plugin(name)
failed_plugins.append(name)
continue
self.instances[name] = instance
for event in instance.handlers:
if event not in self.listening_plugins:
self.listening_plugins[event] = []
self.listening_plugins[event].append(name)
self.refresh_order()
return failed_plugins
def reload_plugin(self, name: str):
name = name.upper()
if name in self.instances:
for event in self.listening_plugins:
if name in self.listening_plugins[event]:
self.listening_plugins[event].remove(name)
del self.instances[name]
self.activate_plugins()
return True
return False
def load_plugins(self):
self.load_config()
self.scan_plugins()
# 加载全量插件配置
self._load_all_config()
pconf = self.pconf
logger.debug("plugins.json config={}".format(pconf))
for name, plugin in pconf["plugins"].items():
if name.upper() not in self.plugins:
logger.error("Plugin %s not found, but found in plugins.json" % name)
self.activate_plugins()
def emit_event(self, e_context: EventContext, *args, **kwargs):
if e_context.event in self.listening_plugins:
for name in self.listening_plugins[e_context.event]:
if self.plugins[name].enabled and e_context.action == EventAction.CONTINUE:
logger.debug("Plugin %s triggered by event %s" % (name, e_context.event))
instance = self.instances[name]
instance.handlers[e_context.event](e_context, *args, **kwargs)
if e_context.is_break():
e_context["breaked_by"] = name
logger.debug("Plugin %s breaked event %s" % (name, e_context.event))
return e_context
def set_plugin_priority(self, name: str, priority: int):
name = name.upper()
if name not in self.plugins:
return False
if self.plugins[name].priority == priority:
return True
self.plugins[name].priority = priority
self.plugins._update_heap(name)
rawname = self.plugins[name].name
self.pconf["plugins"][rawname]["priority"] = priority
self.pconf["plugins"]._update_heap(rawname)
self.save_config()
self.refresh_order()
return True
def enable_plugin(self, name: str):
name = name.upper()
if name not in self.plugins:
return False, "插件不存在"
if not self.plugins[name].enabled:
self.plugins[name].enabled = True
rawname = self.plugins[name].name
self.pconf["plugins"][rawname]["enabled"] = True
self.save_config()
failed_plugins = self.activate_plugins()
if name in failed_plugins:
return False, "插件开启失败"
return True, "插件已开启"
return True, "插件已开启"
def disable_plugin(self, name: str):
name = name.upper()
if name not in self.plugins:
return False
if self.plugins[name].enabled:
self.plugins[name].enabled = False
rawname = self.plugins[name].name
self.pconf["plugins"][rawname]["enabled"] = False
self.save_config()
return True
return True
def list_plugins(self):
return self.plugins
def install_plugin(self, repo: str):
try:
import common.package_manager as pkgmgr
pkgmgr.check_dulwich()
except Exception as e:
logger.error("Failed to install plugin, {}".format(e))
return False, "无法导入dulwich,安装插件失败"
import re
from dulwich import porcelain
logger.info("clone git repo: {}".format(repo))
match = re.match(r"^(https?:\/\/|git@)([^\/:]+)[\/:]([^\/:]+)\/(.+).git$", repo)
if not match:
try:
with open("./plugins/source.json", "r", encoding="utf-8") as f:
source = json.load(f)
if repo in source["repo"]:
repo = source["repo"][repo]["url"]
match = re.match(r"^(https?:\/\/|git@)([^\/:]+)[\/:]([^\/:]+)\/(.+).git$", repo)
if not match:
return False, "安装插件失败,source中的仓库地址不合法"
else:
return False, "安装插件失败,仓库地址不合法"
except Exception as e:
logger.error("Failed to install plugin, {}".format(e))
return False, "安装插件失败,请检查仓库地址是否正确"
dirname = os.path.join("./plugins", match.group(4))
try:
repo = porcelain.clone(repo, dirname, checkout=True)
if os.path.exists(os.path.join(dirname, "requirements.txt")):
logger.info("detect requirements.txt,installing...")
pkgmgr.install_requirements(os.path.join(dirname, "requirements.txt"))
return True, "安装插件成功,请使用 #scanp 命令扫描插件或重启程序,开启前请检查插件是否需要配置"
except Exception as e:
logger.error("Failed to install plugin, {}".format(e))
return False, "安装插件失败," + str(e)
def update_plugin(self, name: str):
try:
import common.package_manager as pkgmgr
pkgmgr.check_dulwich()
except Exception as e:
logger.error("Failed to install plugin, {}".format(e))
return False, "无法导入dulwich,更新插件失败"
from dulwich import porcelain
name = name.upper()
if name not in self.plugins:
return False, "插件不存在"
if name in [
"HELLO",
"GODCMD",
"ROLE",
"TOOL",
"BDUNIT",
"BANWORDS",
"FINISH",
"DUNGEON",
]:
return False, "预置插件无法更新,请更新主程序仓库"
dirname = self.plugins[name].path
try:
porcelain.pull(dirname, "origin")
if os.path.exists(os.path.join(dirname, "requirements.txt")):
logger.info("detect requirements.txt,installing...")
pkgmgr.install_requirements(os.path.join(dirname, "requirements.txt"))
return True, "更新插件成功,请重新运行程序"
except Exception as e:
logger.error("Failed to update plugin, {}".format(e))
return False, "更新插件失败," + str(e)
def uninstall_plugin(self, name: str):
name = name.upper()
if name not in self.plugins:
return False, "插件不存在"
if name in self.instances:
self.disable_plugin(name)
dirname = self.plugins[name].path
try:
import shutil
shutil.rmtree(dirname)
rawname = self.plugins[name].name
for event in self.listening_plugins:
if name in self.listening_plugins[event]:
self.listening_plugins[event].remove(name)
del self.plugins[name]
del self.pconf["plugins"][rawname]
self.loaded[dirname] = None
self.save_config()
return True, "卸载插件成功"
except Exception as e:
logger.error("Failed to uninstall plugin, {}".format(e))
return False, "卸载插件失败,请手动删除文件夹完成卸载," + str(e)
|