ElapsingDreams/LiteyukiBot
1
0
Fork 0
forked from bot/app
Code Pull Requests Activity

feat: 添加插件启用和停用功能

Browse Source
This commit is contained in:
远野千束
2024-03-22 07:44:41 +08:00
parent 2711d8844b
commit 73b593ff98
14 changed files with 335 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/plugins/liteyuki_plugin_npm/__init__.py
View File

@ -2,6 +2,7 @@ from nonebot.plugin import PluginMetadata
from .manager import *
from .installer import *
from .helper import *
from .permission import *
__author__ = "snowykami"
__plugin_meta__ = PluginMetadata(
@ -16,6 +17,8 @@ __plugin_meta__ = PluginMetadata(
type="application",
homepage="https://github.com/snowykami/LiteyukiBot",
extra={
"liteyuki_plugin": True,
"liteyuki": True,
"toggleable" : False,
"default_enable" : True,
}
)

96
src/plugins/liteyuki_plugin_npm/common.py
View File

@ -1,8 +1,102 @@
import json
from typing import Optional
import aiofiles
import nonebot.plugin
from src.utils.data import Database, LiteModel
from src.utils.data_manager import InstalledPlugin, plugin_db
from src.utils.data_manager import GroupChat, InstalledPlugin, User, group_db, plugin_db, user_db
from src.utils.typing import T_MessageEvent
LNPM_COMMAND_START = "lnpm"
class PluginTag(LiteModel):
label: str
color: str = '#000000'
class StorePlugin(LiteModel):
name: str
desc: str
module_name: str
project_link: str = ''
homepage: str = ''
author: str = ''
type: str | None = None
version: str | None = ''
time: str = ''
tags: list[PluginTag] = []
is_official: bool = False
async def get_store_plugin(plugin_module_name: str) -> Optional[StorePlugin]:
"""
获取插件信息
Args:
plugin_module_name (str): 插件模块名
Returns:
Optional[StorePlugin]: 插件信息
"""
async with aiofiles.open("data/liteyuki/plugins.json", "r", encoding="utf-8") as f:
plugins: list[StorePlugin] = [StorePlugin(**pobj) for pobj in json.loads(await f.read())]
for plugin in plugins:
if plugin.module_name == plugin_module_name:
return plugin
return None
def get_plugin_default_enable(plugin_module_name: str) -> bool:
"""
获取插件默认启用状态，由插件定义，不存在则默认为启用
Args:
plugin_module_name (str): 插件模块名
Returns:
bool: 插件默认状态
"""
return (nonebot.plugin.get_plugin(plugin_module_name).metadata.extra.get('default_enable', True)
if nonebot.plugin.get_plugin(plugin_module_name) and nonebot.plugin.get_plugin(plugin_module_name).metadata else True) \
if nonebot.plugin.get_plugin(plugin_module_name) else False
def get_plugin_current_enable(event: T_MessageEvent, plugin_module_name: str) -> bool:
"""
获取插件当前启用状态
Args:
plugin_module_name (str): 插件模块名
Returns:
bool: 插件当前状态
"""
if event.message_type == "group":
session: GroupChat = group_db.first(GroupChat, 'group_id = ?', event.group_id, default=GroupChat(group_id=event.group_id))
else:
session: User = user_db.first(User, 'user_id = ?', event.user_id, default=User(user_id=event.user_id))
# 默认停用插件在启用列表内表示启用
# 默认停用插件不在启用列表内表示停用
# 默认启用插件在停用列表内表示停用
# 默认启用插件不在停用列表内表示启用
default_enable = get_plugin_default_enable(plugin_module_name)
if default_enable:
return plugin_module_name not in session.disabled_plugins
else:
return plugin_module_name in session.enabled_plugins
def get_plugin_can_be_toggle(plugin_module_name: str) -> bool:
"""
获取插件是否可以被启用/停用
Args:
plugin_module_name (str): 插件模块名
Returns:
bool: 插件是否可以被启用/停用
"""
return nonebot.plugin.get_plugin(plugin_module_name).metadata.extra.get('toggleable', True) \
if nonebot.plugin.get_plugin(plugin_module_name) and nonebot.plugin.get_plugin(plugin_module_name).metadata else True

82
src/plugins/liteyuki_plugin_npm/installer.py
View File

@ -22,7 +22,7 @@ from src.utils.data_manager import InstalledPlugin
npm_alc = on_alconna(
Alconna(
["lnpm", "插件管理"],
["npm", "插件"],
Subcommand(
"update",
alias=["u"],
@ -42,29 +42,17 @@ npm_alc = on_alconna(
Args["plugin_name", str],
alias=["rm", "移除", "卸载"],
),
Subcommand(
"list",
alias=["l", "ls", "列表"],
)
),
permission=SUPERUSER
)
class PluginTag(LiteModel):
label: str
color: str = '#000000'
class StorePlugin(LiteModel):
name: str
desc: str
module_name: str
project_link: str = ''
homepage: str = ''
author: str = ''
type: str | None = None
version: str | None = ''
time: str = ''
tags: list[PluginTag] = []
is_official: bool = False
@npm_alc.handle()
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
@ -88,12 +76,15 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
if len(rs):
reply = f"{ulang.get('npm.search_result')} | {ulang.get('npm.total', TOTAL=len(rs))}\n***"
for plugin in rs[:min(max_show, len(rs))]:
btn_install = md.button(ulang.get('npm.install'), 'lnpm install %s' % plugin.module_name)
btn_install = md.button(ulang.get('npm.install'), 'npm install %s' % plugin.module_name)
link_page = md.link(ulang.get('npm.homepage'), plugin.homepage)
link_pypi = md.link(ulang.get('npm.pypi'), plugin.homepage)
reply += (f"\n{btn_install} **{plugin.name}**\n"
f"\n > **{plugin.desc}**\n"
f"\n > {ulang.get('npm.author')}: {plugin.author} {link_page}\n\n***\n")
reply += (f"\n# **{plugin.name}**\n"
f"\n> **{plugin.desc}**\n"
f"\n> {ulang.get('npm.author')}: {plugin.author}"
f"\n> *{md.escape(plugin.module_name)}*"
f"\n> {btn_install} {link_page} {link_pypi}\n\n***\n")
if len(rs) > max_show:
reply += f"\n{ulang.get('npm.too_many_results', HIDE_NUM=len(rs) - max_show)}"
else:
@ -102,27 +93,43 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
elif result.subcommands.get("install"):
plugin_module_name: str = result.subcommands["install"].args.get("plugin_name")
store_plugin = await get_store_plugin(plugin_module_name)
await npm_alc.send(ulang.get("npm.installing", NAME=plugin_module_name))
r, log = npm_install(plugin_module_name)
log = log.replace("\\", "/")
if not store_plugin:
await npm_alc.finish(ulang.get("npm.plugin_not_found", NAME=plugin_module_name))
homepage_btn = md.button(ulang.get('npm.homepage'), store_plugin.homepage)
if r:
nonebot.load_plugin(plugin_module_name) # 加载插件
r_load = nonebot.load_plugin(plugin_module_name) # 加载插件
installed_plugin = InstalledPlugin(module_name=plugin_module_name) # 构造插件信息模型
store_plugin = await get_store_plugin(plugin_module_name) # 获取商店中的插件信息
found_in_db_plugin = plugin_db.first(InstalledPlugin, "module_name = ?", plugin_module_name) # 查询数据库中是否已经安装
if found_in_db_plugin is None:
plugin_db.save(installed_plugin)
info = ulang.get('npm.install_success', NAME=store_plugin.name).replace('_', r'\\_') # markdown转义
if r_load:
if found_in_db_plugin is None:
plugin_db.save(installed_plugin)
info = ulang.get('npm.install_success', NAME=store_plugin.name).replace('_', r'\\_') # markdown转义
await send_markdown(
f"{info}\n\n"
f"```\n{log}\n```",
bot,
event=event
)
else:
await npm_alc.finish(ulang.get('npm.plugin_already_installed', NAME=store_plugin.name))
else:
info = ulang.get('npm.load_failed', NAME=plugin_module_name, HOMEPAGE=homepage_btn).replace('_', r'\\_')
await send_markdown(
f"{info}\n\n"
f"```\n{log}\n```",
f"```\n{log}\n```\n",
bot,
event=event
)
else:
await npm_alc.finish(ulang.get('npm.plugin_already_installed', NAME=store_plugin.name))
else:
info = ulang.get('npm.install_failed', NAME=plugin_module_name).replace('_', r'\\_')
info = ulang.get('npm.install_failed', NAME=plugin_module_name, HOMEPAGE=homepage_btn).replace('_', r'\\_')
await send_markdown(
f"{info}\n\n"
f"```\n{log}\n```",
@ -192,22 +199,7 @@ async def npm_search(keywords: list[str]) -> list[StorePlugin]:
return results
async def get_store_plugin(plugin_module_name: str) -> Optional[StorePlugin]:
"""
获取插件信息
Args:
plugin_module_name (str): 插件模块名
Returns:
Optional[StorePlugin]: 插件信息
"""
async with aiofiles.open("data/liteyuki/plugins.json", "r", encoding="utf-8") as f:
plugins: list[StorePlugin] = [StorePlugin(**pobj) for pobj in json.loads(await f.read())]
for plugin in plugins:
if plugin.module_name == plugin_module_name:
return plugin
return None
def npm_install(plugin_module_name) -> tuple[bool, str]:

116
src/plugins/liteyuki_plugin_npm/manager.py
View File

@ -1,41 +1,131 @@
import nonebot.plugin
from nonebot import on_command
from nonebot.internal.matcher import Matcher
from nonebot.message import run_preprocessor
from nonebot.permission import SUPERUSER
from nonebot_plugin_alconna import on_alconna, Alconna, Args, Arparma
from src.utils.data_manager import InstalledPlugin, plugin_db
from src.utils.data_manager import GroupChat, InstalledPlugin, User, group_db, plugin_db, user_db
from src.utils.message import Markdown as md, send_markdown
from src.utils.permission import GROUP_ADMIN, GROUP_OWNER
from src.utils.typing import T_Bot, T_MessageEvent
from src.utils.language import get_user_lang
from .common import get_plugin_can_be_toggle, get_plugin_current_enable, get_plugin_default_enable
from .installer import get_store_plugin
list_plugins = on_command("list-plugin", aliases={"列出插件", "插件列表"}, priority=0)
toggle_plugin = on_command("enable-plugin", aliases={"启用插件", "停用插件", "disable-plugin"}, priority=0, permission=SUPERUSER)
# toggle_plugin = on_command("enable-plugin", aliases={"启用插件", "停用插件", "disable-plugin"}, priority=0)
toggle_plugin = on_alconna(
Alconna(
['enable-plugin', 'disable-plugin'],
Args['plugin_name', str]['global', bool, False],
)
)
@list_plugins.handle()
async def _(event: T_MessageEvent, bot: T_Bot):
lang = get_user_lang(str(event.user_id))
reply = f"# {lang.get('npm.loaded_plugins')} | {lang.get('npm.total', TOTAL=len(nonebot.get_loaded_plugins()))} \n***"
reply = f"# {lang.get('npm.loaded_plugins')} | {lang.get('npm.total', TOTAL=len(nonebot.get_loaded_plugins()))} \n***\n"
for plugin in nonebot.get_loaded_plugins():
# 检查是否有 metadata 属性
btn_help = md.button(lang.get('npm.help'), f'help {plugin.name}', False)
reply += f"\n{btn_help} "
# 添加帮助按钮
btn_usage = md.button(lang.get('npm.usage'), f'help {plugin.name}', False)
store_plugin = await get_store_plugin(plugin.module_name)
if store_plugin:
btn_homepage = md.link(lang.get('npm.homepage'), store_plugin.homepage)
elif plugin.metadata and plugin.metadata.extra.get('liteyuki'):
btn_homepage = md.link(lang.get('npm.homepage'), "https://github.com/snowykami/LiteyukiBot")
else:
btn_homepage = lang.get('npm.homepage')
if plugin.metadata:
reply += (f"**{plugin.metadata.name}**\n"
reply += (f"\n**{md.escape(plugin.metadata.name)}**\n"
f"\n > {plugin.metadata.description}")
else:
reply += (f"**{plugin.name}**\n"
reply += (f"**{md.escape(plugin.name)}**\n"
f"\n > {lang.get('npm.no_description')}")
# if await GROUP_ADMIN(bot=bot, event=event) or await GROUP_OWNER(bot=bot, event=event) or await SUPERUSER(bot=bot, event=event):
reply += f"\n > {btn_usage} {btn_homepage}"
if await GROUP_ADMIN(bot, event) or await GROUP_OWNER(bot, event) or await SUPERUSER(bot, event):
btn_enable = md.button(lang.get('npm.enable'), f'enable-plugin {plugin.module_name}')
btn_disable = md.button(lang.get('npm.disable'), f'disable-plugin {plugin.module_name}')
reply += f"\n > {btn_enable} {btn_disable}"
# 添加启用/停用插件按钮
btn_toggle = lang.get('npm.disable') if plugin.metadata and not plugin.metadata.extra.get('toggleable') \
else md.button(lang.get('npm.disable'), f'enable-plugin {plugin.module_name}')
reply += f" {btn_toggle}"
if await SUPERUSER(bot, event):
plugin_in_database = plugin_db.first(InstalledPlugin, 'module_name = ?', plugin.module_name)
# 添加移除插件
btn_remove = (
md.button(lang.get('npm.uninstall'), f'lnpm remove {plugin.module_name}')) if plugin_in_database else lang.get(
md.button(lang.get('npm.uninstall'), f'npm remove {plugin.module_name}')) if plugin_in_database else lang.get(
'npm.uninstall')
reply += f" {btn_remove}"
btn_toggle_global = lang.get('npm.disable') if plugin.metadata and not plugin.metadata.extra.get('toggleable') \
else md.button(lang.get('npm.disable_global'), f'disable-plugin {plugin.module_name} global')
reply += f" {btn_remove} {btn_toggle_global}"
reply += "\n\n***\n"
await send_markdown(reply, bot, event=event)
@toggle_plugin.handle()
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
# 判断会话类型
ulang = get_user_lang(str(event.user_id))
plugin_module_name = result.args.get("plugin_name")
toggle = result.header_result == 'enable-plugin' # 判断是启用还是停用
current_enable = get_plugin_current_enable(event, plugin_module_name) # 获取插件当前状态
default_enable = get_plugin_default_enable(plugin_module_name) # 获取插件默认状态
can_be_toggled = get_plugin_can_be_toggle(plugin_module_name) # 获取插件是否可以被启用/停用
if not can_be_toggled:
await toggle_plugin.finish(ulang.get("npm.plugin_cannot_be_toggled", NAME=plugin_module_name))
if current_enable == toggle:
await toggle_plugin.finish(
ulang.get("npm.plugin_already", NAME=plugin_module_name, STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable")))
if event.message_type == "private":
session = user_db.first(User, "user_id = ?", event.user_id, default=User(user_id=event.user_id))
else:
if await GROUP_ADMIN(bot, event) or await GROUP_OWNER(bot, event) or await SUPERUSER(bot, event):
session = group_db.first(GroupChat, "group_id = ?", event.group_id, default=GroupChat(group_id=event.group_id))
else:
return
# 启用 已停用的默认启用插件 将其从停用列表移除
# 启用 已停用的默认停用插件 将其放到启用列表
# 停用 已启用的默认启用插件 将其放到停用列表
# 停用 已启用的默认停用插件 将其从启用列表移除
try:
if toggle:
if default_enable:
session.disabled_plugins.remove(plugin_module_name)
else:
session.enabled_plugins.append(plugin_module_name)
else:
if default_enable:
session.disabled_plugins.append(plugin_module_name)
else:
session.enabled_plugins.remove(plugin_module_name)
except Exception as e:
await toggle_plugin.finish(
ulang.get(
"npm.toggle_failed",
NAME=plugin_module_name,
STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"),
ERROR=str(e))
)
if event.message_type == "private":
user_db.save(session)
else:
group_db.save(session)
@run_preprocessor
async def _(event: T_MessageEvent, matcher: Matcher):
plugin = matcher.plugin
nonebot.logger.info(f"Plugin: {plugin.module_name}")