Genshin-bots/gsuid_core
1
0
Fork 0
mirror of https://github.com/Genshin-bots/gsuid_core.git synced 2025-05-07 20:45:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

✨ 支持插件热重载, 更新插件无需重启, 新增core手动重载插件

Browse Source
This commit is contained in:
KimigaiiWuyi 2025-02-07 02:22:37 +08:00
parent 00f17c417b
commit 0bc16c06b9
6 changed files with 192 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
gsuid_core/buildin_plugins/core_command/core_restart/restart.py
View File

@ -5,7 +5,7 @@ import platform
import subprocess
from pathlib import Path
from gsuid_core.server import check_start_tool
from gsuid_core.utils.plugins_update.utils import check_start_tool
from gsuid_core.utils.plugins_config.gs_config import core_plugins_config
from ..core_status.command_global_val import save_global_val

10
gsuid_core/buildin_plugins/core_command/core_update/__init__.py
View File

@ -5,6 +5,7 @@ from gsuid_core.sv import SV
from gsuid_core.bot import Bot
from gsuid_core.models import Event
from gsuid_core.logger import logger
from gsuid_core.utils.plugins_update.reload_plugin import reload_plugin
from gsuid_core.utils.plugins_config.gs_config import core_plugins_config
from gsuid_core.utils.plugins_update._plugins import (
run_install,
@ -17,6 +18,15 @@ from gsuid_core.utils.plugins_update._plugins import (
sv_core_config = SV('Core管理', pm=0)
@sv_core_config.on_prefix(('core手动重载插件'))
async def send_core_reload_msg(bot: Bot, ev: Event):
plugin_name = ev.text.strip()
logger.info(f'🔔 开始执行 [重载] {plugin_name}')
await bot.send(f'🔔 正在尝试重载插件{plugin_name}...')
retcode = reload_plugin(plugin_name)
await bot.send(retcode)
@sv_core_config.on_fullmatch(('core更新', 'core强制更新'), block=True)
async def send_core_update_msg(bot: Bot, ev: Event):
logger.info('开始执行[更新] 早柚核心')

118
gsuid_core/server.py
View File

@ -4,7 +4,8 @@ import asyncio
import importlib
import subprocess
from pathlib import Path
from typing import Dict, List, Callable
from types import ModuleType
from typing import Dict, List, Union, Callable
import toml
import pkg_resources
@ -12,7 +13,7 @@ from fastapi import WebSocket
from gsuid_core.bot import _Bot
from gsuid_core.logger import logger
from gsuid_core.utils.plugins_update._plugins import check_start_tool
from gsuid_core.utils.plugins_update.utils import check_start_tool
from gsuid_core.utils.plugins_config.gs_config import core_plugins_config
auto_install_dep: bool = core_plugins_config.get_config('AutoInstallDep').data
@ -60,6 +61,64 @@ class GsServer:
self.active_bot: Dict[str, _Bot] = {}
self.is_initialized = True
def load_plugin(self, plugin: Union[str, Path]):
if isinstance(plugin, str):
plugin = PLUGIN_PATH / plugin
if not plugin.exists():
logger.warning(f'[更新] ❌ 插件{plugin.name}不存在!')
return f'❌ 插件{plugin.name}不存在!'
plugin_parent = plugin.parent.name
if plugin.stem.startswith('_'):
return f'插件{plugin.name}包含"_", 跳过加载!'
# 如果发现文件夹，则视为插件包
logger.trace('===============')
logger.debug(f'🔹 导入{plugin.stem}中...')
logger.trace('===============')
try:
if plugin.is_dir():
plugin_path = plugin / '__init__.py'
plugins_path = plugin / '__full__.py'
nest_path = plugin / '__nest__.py'
src_path = plugin / plugin.stem
# 如果文件夹内有__full_.py，则视为插件包合集
sys.path.append(str(plugin_path.parents))
if plugins_path.exists():
module_list = self.load_dir_plugins(plugin, plugin_parent)
elif nest_path.exists() or src_path.exists():
path = nest_path.parent / plugin.name
pyproject = plugin / 'pyproject.toml'
if pyproject.exists:
check_pyproject(pyproject)
if path.exists():
module_list = self.load_dir_plugins(
path, plugin_parent, True
)
# 如果文件夹内有__init_.py，则视为单个插件包
elif plugin_path.exists():
module_list = [
importlib.import_module(
f'{plugin_parent}.{plugin.name}.__init__'
)
]
# 如果发现单文件，则视为单文件插件
elif plugin.suffix == '.py':
module_list = [
importlib.import_module(
f'{plugin_parent}.{plugin.name[:-3]}'
)
]
'''导入成功'''
logger.success(f'✅ 插件{plugin.stem}导入成功!')
return module_list
except Exception as e: # noqa
exception = sys.exc_info()
logger.opt(exception=exception).error(f'加载插件时发生错误: {e}')
logger.warning(f'❌ 插件{plugin.name}加载失败')
return f'❌ 插件{plugin.name}加载失败'
def load_plugins(self):
logger.info('[GsCore] 开始加载插件...')
get_installed_dependencies()
@ -71,58 +130,22 @@ class GsServer:
# 遍历插件文件夹内所有文件
for plugin in plug_path_list:
plugin_parent = plugin.parent.name
if plugin.stem.startswith('_'):
continue
# 如果发现文件夹，则视为插件包
logger.trace('===============')
logger.debug(f'导入{plugin.stem}中...')
logger.trace('===============')
try:
if plugin.is_dir():
plugin_path = plugin / '__init__.py'
plugins_path = plugin / '__full__.py'
nest_path = plugin / '__nest__.py'
src_path = plugin / plugin.stem
# 如果文件夹内有__full_.py，则视为插件包合集
sys.path.append(str(plugin_path.parents))
if plugins_path.exists():
self.load_dir_plugins(plugin, plugin_parent)
elif nest_path.exists() or src_path.exists():
path = nest_path.parent / plugin.name
pyproject = plugin / 'pyproject.toml'
if pyproject.exists:
check_pyproject(pyproject)
if path.exists():
self.load_dir_plugins(path, plugin_parent, True)
# 如果文件夹内有__init_.py，则视为单个插件包
elif plugin_path.exists():
importlib.import_module(
f'{plugin_parent}.{plugin.name}.__init__'
)
# 如果发现单文件，则视为单文件插件
elif plugin.suffix == '.py':
importlib.import_module(
f'{plugin_parent}.{plugin.name[:-3]}'
)
'''导入成功'''
logger.success(f'插件{plugin.stem}导入成功!')
except Exception as e: # noqa
exception = sys.exc_info()
logger.opt(exception=exception).error(
f'加载插件时发生错误: {e}'
)
logger.warning(f'插件{plugin.name}加载失败')
self.load_plugin(plugin)
logger.info('[GsCore] 插件加载完成!')
def load_dir_plugins(
self, plugin: Path, plugin_parent: str, nest: bool = False
):
) -> List[ModuleType]:
module_list = []
init_path = plugin / '__init__.py'
name = plugin.name
if init_path.exists():
if str(init_path.parents) not in sys.path:
sys.path.append(str(init_path.parents))
importlib.import_module(f'{plugin_parent}.{name}.{name}.__init__')
module = importlib.import_module(
f'{plugin_parent}.{name}.{name}.__init__'
)
module_list.append(module)
for sub_plugin in plugin.iterdir():
if sub_plugin.is_dir():
@ -134,7 +157,8 @@ class GsServer:
_p = f'{plugin_parent}.{name}.{name}.{sub_plugin.name}'
else:
_p = f'{plugin_parent}.{name}.{sub_plugin.name}'
importlib.import_module(f'{_p}')
module_list.append(importlib.import_module(f'{_p}'))
return module_list
async def connect(self, websocket: WebSocket, bot_id: str) -> _Bot:
await websocket.accept()

72
gsuid_core/utils/plugins_update/_plugins.py
View File

@ -6,7 +6,6 @@ from pathlib import Path
from typing import Dict, List, Union, Optional
from concurrent.futures import ThreadPoolExecutor
import psutil
import aiohttp
from git.repo import Repo
from git.exc import GitCommandError, NoSuchPathError, InvalidGitRepositoryError
@ -14,74 +13,15 @@ from git.exc import GitCommandError, NoSuchPathError, InvalidGitRepositoryError
from gsuid_core.logger import logger
from gsuid_core.utils.plugins_config.gs_config import core_plugins_config
from .utils import check_start_tool
from .reload_plugin import reload_plugin
from .api import CORE_PATH, PLUGINS_PATH, plugins_lib
plugins_list: Dict[str, Dict[str, str]] = {}
start_venv: str = core_plugins_config.get_config('StartVENV').data
is_install_dep = core_plugins_config.get_config('AutoInstallDep').data
def get_command_chain() -> List[str]:
cmd_chain = []
process = psutil.Process()
while process:
try:
cmd_chain.extend(process.cmdline())
process = process.parent()
except Exception as e:
logger.warning(f'获取命令链失败...{e}')
break
return cmd_chain
def check_start_tool(is_pip: bool = False):
command_chain = get_command_chain()
command_chain = [command.lower() for command in command_chain]
command_chain_str = ' '.join(command_chain)
logger.debug(f'[检测启动工具] 命令链: {command_chain}')
PDM = 'pdm'
POETRY = 'poetry'
UV = 'uv'
OTHER = start_venv.strip()
PYTHON = 'python'
if OTHER == 'auto':
OTHER = PYTHON
if is_pip:
PIP = ' run python -m pip'
PDM += PIP
POETRY += ' run pip'
UV += PIP
PYTHON += ' -m pip'
if OTHER == 'python' or OTHER == 'auto':
OTHER = 'python -m pip'
else:
OTHER += PIP
if start_venv == 'auto':
if 'pdm' in command_chain_str:
command = PDM
elif 'poetry' in command_chain_str:
command = POETRY
elif 'uv' in command_chain or 'uv.exe' in command_chain_str:
command = UV
else:
command = PYTHON
elif start_venv == 'pdm':
command = PDM
elif start_venv == 'poetry':
command = POETRY
elif start_venv == 'uv':
command = UV
else:
command = OTHER
return command
async def check_plugin_exist(name: str):
name = name.lower()
if name in ['core_command', 'gs_test']:
@ -532,6 +472,8 @@ def update_from_git(
break
else:
log_list.append(f'✅插件 {plugin_name} 本次无更新内容！')
if plugin_name != '早柚核心':
reload_plugin(plugin_name)
return log_list
@ -562,7 +504,9 @@ async def update_plugins(
break
log_list = await update_from_git_in_tread(
level, plugin_name, log_key, log_limit
level,
plugin_name,
log_key,
log_limit,
)
return log_list
return log_list

34
gsuid_core/utils/plugins_update/reload_plugin.py Normal file
View File

@ -0,0 +1,34 @@
import importlib
from gsuid_core.sv import SL
from gsuid_core.gss import gss
from gsuid_core.logger import logger
def reload_plugin(plugin_name: str):
logger.info(f'🔔 正在重载插件{plugin_name}...')
del_k = []
del_v = []
for sv_name in SL.lst:
sv = SL.lst[sv_name]
if sv.plugins.name == plugin_name:
del_k.append(sv_name)
if sv.plugins not in del_v:
del_v.append(sv.plugins)
for k in del_k:
del SL.lst[k]
for v in del_v:
del SL.detail_lst[v]
del SL.plugins[plugin_name]
retcode = gss.load_plugin(plugin_name)
if isinstance(retcode, str):
logger.info(f'❌ 重载插件{plugin_name}失败...')
return retcode
else:
for module in retcode:
importlib.reload(module)
logger.info(f'✨ 已重载插件{plugin_name}')
return f'✨ 已重载插件{plugin_name}!'

68
gsuid_core/utils/plugins_update/utils.py Normal file
View File

@ -0,0 +1,68 @@
from typing import List
import psutil
from gsuid_core.logger import logger
from gsuid_core.utils.plugins_config.gs_config import core_plugins_config
start_venv: str = core_plugins_config.get_config('StartVENV').data
def get_command_chain() -> List[str]:
cmd_chain = []
process = psutil.Process()
while process:
try:
cmd_chain.extend(process.cmdline())
process = process.parent()
except Exception as e:
logger.warning(f'获取命令链失败...{e}')
break
return cmd_chain
def check_start_tool(is_pip: bool = False):
command_chain = get_command_chain()
command_chain = [command.lower() for command in command_chain]
command_chain_str = ' '.join(command_chain)
logger.debug(f'[检测启动工具] 命令链: {command_chain}')
PDM = 'pdm'
POETRY = 'poetry'
UV = 'uv'
OTHER = start_venv.strip()
PYTHON = 'python'
if OTHER == 'auto':
OTHER = PYTHON
if is_pip:
PIP = ' run python -m pip'
PDM += PIP
POETRY += ' run pip'
UV += PIP
PYTHON += ' -m pip'
if OTHER == 'python' or OTHER == 'auto':
OTHER = 'python -m pip'
else:
OTHER += PIP
if start_venv == 'auto':
if 'pdm' in command_chain_str:
command = PDM
elif 'poetry' in command_chain_str:
command = POETRY
elif 'uv' in command_chain or 'uv.exe' in command_chain_str:
command = UV
else:
command = PYTHON
elif start_venv == 'pdm':
command = PDM
elif start_venv == 'poetry':
command = POETRY
elif start_venv == 'uv':
command = UV
else:
command = OTHER
return command