Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2025-12-15 06:12:53 +08:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

perf👌: 完善插件管理api

Browse Source
This commit is contained in:
HibiKier 2024-01-08 05:39:44 +08:00
parent 87a39bf033
commit 8acbe0bb1c
5 changed files with 129 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
plugins/web_ui/__init__.py
View File

@ -30,6 +30,7 @@ gConfig.add_plugin_config("web-ui", "password", None, name="web-ui", help_="前
BaseApiRouter = APIRouter(prefix="/zhenxun/api")
BaseApiRouter.include_router(auth_router)
BaseApiRouter.include_router(main_router)
BaseApiRouter.include_router(manage_router)
@ -37,12 +38,24 @@ BaseApiRouter.include_router(database_router)
BaseApiRouter.include_router(plugin_router)
WsApiRouter = APIRouter(prefix="/zhenxun/socket")
WsApiRouter.include_router(ws_log_routes)
WsApiRouter.include_router(status_routes)
@driver.on_startup
def _():
try:
loop = asyncio.get_running_loop()
def log_sink(message: str):
async def log_sink(message: str):
loop = None
if not loop:
try:
loop = asyncio.get_running_loop()
except Exception as e:
logger.warning('Web Ui log_sink', e=e)
if not loop:
loop = asyncio.new_event_loop()
loop.create_task(LOG_STORAGE.add(message.rstrip("\n")))
logger_.add(
@ -51,8 +64,7 @@ def _():
app: FastAPI = nonebot.get_app()
app.include_router(BaseApiRouter)
app.include_router(ws_log_routes)
app.include_router(status_routes)
app.include_router(WsApiRouter)
logger.info("<g>API启动成功</g>", "Web UI")
except Exception as e:
logger.error("<g>API启动失败</g>", "Web UI", e=e)

113
plugins/web_ui/api/tabs/plugin_manage/__init__.py
View File

@ -1,3 +1,4 @@
import re
from typing import List, Optional
import cattrs
@ -10,7 +11,15 @@ from utils.manager.models import PluginData, PluginSetting, PluginType
from ....base_model import Result
from ....utils import authentication
from .model import PluginCount, PluginInfo, PluginSwitch, UpdateConfig, UpdatePlugin
from .model import (
PluginConfig,
PluginCount,
PluginDetail,
PluginInfo,
PluginSwitch,
UpdateConfig,
UpdatePlugin,
)
router = APIRouter()
@ -36,7 +45,7 @@ def _(
plugin_info = PluginInfo(
module=module,
plugin_name=plugin_data.name,
default_switch=getattr(setting, "default_status", False),
default_status=getattr(setting, "default_status", False),
limit_superuser=getattr(setting, "limit_superuser", False),
cost_gold=getattr(setting, "cost_gold", 0),
menu_type=menu_type_,
@ -67,7 +76,7 @@ def _() -> Result:
plugin_count.other += 1
return Result.ok(plugin_count)
@router.post("/update_plugins", dependencies=[authentication()], description="更新插件参数")
@router.post("/update_plugin", dependencies=[authentication()], description="更新插件参数")
def _(plugin: UpdatePlugin) -> Result:
"""
修改插件信息
@ -79,16 +88,15 @@ def _(plugin: UpdatePlugin) -> Result:
p2s.default_status = plugin.default_status
p2s.limit_superuser = plugin.limit_superuser
p2s.cost_gold = plugin.cost_gold
p2s.cmd = plugin.cmd.split(",") if plugin.cmd else []
# p2s.cmd = plugin.cmd.split(",") if plugin.cmd else []
p2s.level = plugin.level
if pd := plugin_data_manager.get(module):
menu_lin = None
if len(pd.menu_type) > 1:
menu_lin = pd.menu_type[1]
if len(p2s.plugin_type) > 1:
menu_lin = p2s.menu_type[1]
if menu_lin is not None:
pd.menu_type = (plugin.menu_type, menu_lin)
p2s.plugin_type = (plugin.menu_type, menu_lin)
else:
pd.menu_type = (plugin.menu_type,)
p2s.plugin_type = (plugin.menu_type,)
if pm := plugins_manager.get(module):
if plugin.block_type:
pm.block_type = plugin.block_type
@ -98,34 +106,25 @@ def _(plugin: UpdatePlugin) -> Result:
pm.status = True
plugins2settings_manager.save()
plugins_manager.save()
# 配置项
if plugin.configs and (configs := Config.get(module)):
for key in plugin.configs:
if c := configs.configs.get(key):
value = plugin.configs[key]
if isinstance(c.value, (list, tuple)) or isinstance(
c.default_value, (list, tuple)
):
value = value.split(",")
if c.type and value is not None:
value = cattrs.structure(value, c.type)
Config.set_config(module, key, value)
plugin_data_manager.reload()
except Exception as e:
logger.error("调用API错误", "/update_plugins", e=e)
return Result.fail(f"{type(e)}: {e}")
return Result.ok(info="已经帮你写好啦!")
@router.post("/update_config", dependencies=[authentication()], description="更新配置")
def _(config_list: List[UpdateConfig]) -> Result:
try:
for config in config_list:
if cg := Config.get(config.module):
if c := cg.configs.get(config.key):
if isinstance(c.value, (list, tuple)) or isinstance(
c.default_value, (list, tuple)
):
value = config.value.split(",")
else:
value = config.value
if c.type and value is not None:
value = cattrs.structure(value, c.type)
Config.set_config(config.module, config.key, value)
except Exception as e:
logger.error("调用API错误", "/update_config", e=e)
return Result.fail(f"{type(e)}: {e}")
Config.save(save_simple_data=True)
return Result.ok(info="写入配置项了哦!")
@router.post("/change_switch", dependencies=[authentication()], description="开关插件")
def _(param: PluginSwitch) -> Result:
if pm := plugins_manager.get(param.module):
@ -146,4 +145,54 @@ def _() -> Result:
menu_type = getattr(setting, "plugin_type", ["无"])[0]
if menu_type not in menu_type_list:
menu_type_list.append(menu_type)
return Result.ok(menu_type_list)
return Result.ok(menu_type_list)
@router.get("/get_plugin", dependencies=[authentication()], description="获取插件详情")
def _(module: str) -> Result:
if plugin_data := plugin_data_manager.get(module):
setting = plugin_data.plugin_setting or PluginSetting()
plugin = plugin_data.plugin_status
config_list = []
if config := Config.get(module):
for cfg in config.configs:
type_str = ""
type_inner = None
x = str(config.configs[cfg].type)
r = re.search(r"<class '(.*)'>",str(config.configs[cfg].type))
if r:
type_str = r.group(1)
else:
r = re.search(r"typing\.(.*)\[(.*)\]",str(config.configs[cfg].type))
if r:
type_str = r.group(1)
if type_str:
type_str = type_str.lower()
type_inner = r.group(2)
if type_inner:
type_inner = [x.strip() for x in type_inner.split(",")]
config_list.append(PluginConfig(
module=module,
key=cfg,
value=config.configs[cfg].value,
help=config.configs[cfg].help,
default_value=config.configs[cfg].default_value,
type=type_str,
type_inner=type_inner
))
plugin_info = PluginDetail(
module=module,
plugin_name=plugin_data.name,
default_status=getattr(setting, "default_status", False),
limit_superuser=getattr(setting, "limit_superuser", False),
cost_gold=getattr(setting, "cost_gold", 0),
menu_type=getattr(setting, "plugin_type", ["无"])[0],
version=(plugin.version or 0) if plugin else 0,
level=getattr(setting, "level", 5),
status=plugin.status if plugin else False,
author=plugin.author if plugin else None,
config_list=config_list,
block_type=getattr(plugin, "block_type", None)
)
return Result.ok(plugin_info)
return Result.warning_("未获取到插件详情...")

41
plugins/web_ui/api/tabs/plugin_manage/model.py
View File

@ -3,13 +3,7 @@ from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel
from utils.manager.models import Plugin as PluginManager
from utils.manager.models import (
PluginBlock,
PluginCd,
PluginCount,
PluginSetting,
PluginType,
)
from utils.manager.models import PluginBlock, PluginCd, PluginCount, PluginSetting
from utils.typing import BLOCK_TYPE
@ -50,14 +44,14 @@ class UpdatePlugin(BaseModel):
"""限制超级用户"""
cost_gold: int
"""金币花费"""
cmd: str
"""插件别名"""
menu_type: str
"""插件菜单类型"""
level: int
"""插件所需群权限"""
block_type: BLOCK_TYPE
block_type: Optional[BLOCK_TYPE] = None
"""禁用类型"""
configs: Optional[Dict[str, Any]] = None
"""配置项"""
class PluginInfo(BaseModel):
@ -69,7 +63,7 @@ class PluginInfo(BaseModel):
"""插件名称"""
plugin_name: str
"""插件中文名称"""
default_switch: bool
default_status: bool
"""默认开关"""
limit_superuser: bool
"""限制超级用户"""
@ -85,6 +79,8 @@ class PluginInfo(BaseModel):
"""当前状态"""
author: Optional[str] = None
"""作者"""
block_type: BLOCK_TYPE = None
"""禁用类型"""
class PluginConfig(BaseModel):
@ -93,11 +89,20 @@ class PluginConfig(BaseModel):
"""
module: str
"""模块"""
key: str
"""键"""
value: Any
help: Optional[str]
"""值"""
help: Optional[str] = None
"""帮助"""
default_value: Any
has_type: bool
"""默认值"""
type: Optional[Any] = None
"""值类型"""
type_inner: Optional[List[str]] = None
"""List Tuple等内部类型检验"""
class Plugin(BaseModel):
@ -132,4 +137,12 @@ class PluginCount(BaseModel):
superuser: int = 0
"""超级用户插件"""
other: int = 0
"""其他插件"""
"""其他插件"""
class PluginDetail(PluginInfo):
"""
插件详情
"""
config_list: List[PluginConfig]

3
utils/manager/plugin_data_manager.py
View File

@ -29,3 +29,6 @@ class PluginDataManager(StaticData[PluginData]):
def __getitem__(self, item) -> Optional[PluginData]:
return self._data.get(item)
def reload(self):
pass

1
utils/typing.py
View File

@ -1,3 +1,4 @@
from typing import Literal
BLOCK_TYPE = Literal["all", "private", "group"]
"""禁用类型"""