fix: npm对插件无法启用的bug

feat: 资源包的管理器
This commit is contained in:
snowy 2024-04-07 00:35:53 +08:00
parent 72742d805c
commit 9743868cce
15 changed files with 540 additions and 288 deletions

View File

@ -7,7 +7,7 @@ category: 使用手册
## 基础插件命令
#### 命令前有[S]的表示仅超级用户可用,[O]和[A]分别为群主和群管可用
#### 命令前有[S]的表示仅超级用户可用,[O]和[A]分别为群主和群管可用[P]为私聊可用
### 轻雪`liteyuki`
@ -19,31 +19,43 @@ category: 使用手册
[S]config get [key] # 查询配置项不带key返回配置项列表推荐私聊使用
[S]reload-resources # 重载资源
[S]switch-image-mode # 切换图片模式该功能需要commit:505468b及以后的Lagrange.OneBot在普通图片和Markdown图片之间切换后者更大但有失败的可能
liteyuki-docs # 查看轻雪文档
# 上述两个命令修改的配置项在数据库中保存,但是优先级低于配置文件,如果配置文件中存在相同的配置项,将会使用配置文件中的配置
------
别名: reload-liteyuki 重启轻雪, update-liteyuki 更新轻雪, reload-resources 重载资源, config 配置, set 设置, get 查询
别名: reload-liteyuki 重启轻雪, update-liteyuki 更新轻雪, reload-resources 重载资源, config 配置, set 设置, get 查询,
switch-image-mode 切换图片模式, liteyuki-docs 轻雪文档
```
### 轻雪Nonebot插件管理 `liteyuki_npm`
### 轻雪包管理器 `liteyuki_npm`
```shell
[S]npm update # 更新插件索引
[S]npm install <plugin_name> # 安装插件
[S]npm uninstall <plugin_name> # 卸载插件
[S]npm search <keywords...> # 通过关键词搜索插件
[S]nps update # 更新插件索引
[S]nps install <plugin_name> # 安装插件
[S]nps uninstall <plugin_name> # 卸载插件
[S]nps search <keywords...> # 通过关键词搜索插件
------
别名: npm 插件, update 更新, install 安装, uninstall 卸载, search 搜索
[AOSP]npm enable <plugin_name> # 当前会话启用插件
[AOSP]npm disable <plugin_name> # 当前会话禁用插件
[S]npm enable-global <plugin_name> # 全局启用插件
[S]npm disable-global <plugin_name> # 全局禁用插件
list-plugin [page] [num] # 列出所有插件 page为页数num为每页显示数量
------
[S]rpm list [page] [num] # 列出所有资源包 page为页数num为每页显示数量
[S]rpm load <resource_pack_name> # 加载资源包
[S]rpm unload <resource_pack_name> # 卸载资源包
[S]rpm change <resource_pack_name> # 修改优先级
[S]rpm reload # 重载所有资源包
------
别名: nps 插件商店, npm 插件管理, update 更新, install 安装, uninstall 卸载, search 搜索,
enable 启用, disable 停用, enable-global 全局启用, disable-global 全局停用, list-plugin 列出插件/插件列表,
rpm 资源包, load 加载, unload 卸载, change 更改, reload 重载, list 列表/列出
```
```shell
[SOA]enable <plugin_name> # 启用插件
[SOA]disable <plugin_name> # 禁用插件
[S]enable-global <plugin_name> # 全局启用插件
[S]disable-global <plugin_name> # 全局禁用插件
list-plugin [page] [num] # 列出所有插件 page为页数num为每页显示数量
# 受限于Nonebot的钩子函数目前只能阻断消息事件的传入对于主动推送消息的插件无法将其阻止
------
别名: enable 启用, disable 停用, enable-global 全局启用, disable-global 全局停用, list-plugin 列出插件/插件列表
```
### 轻雪用户管理`liteyuki_user`

View File

@ -6,6 +6,7 @@ import pip
from git import Repo
from nonebot import Bot, require, get_driver
from nonebot.exception import MockApiException
from nonebot.internal.matcher import Matcher
from nonebot.permission import SUPERUSER
from liteyuki.utils.config import config, load_from_yaml
@ -23,71 +24,23 @@ driver = get_driver()
markdown_image = common_db.first(StoredConfig(), default=StoredConfig()).config.get("markdown_image", False)
liteyuki = on_alconna(
@on_alconna(
command=Alconna(
"liteecho",
),
permission=SUPERUSER
)
).handle()
async def _(bot: T_Bot, matcher: Matcher):
await matcher.finish(f"Hello, Liteyuki!\nBot {bot.self_id}")
update_liteyuki = on_alconna(
@on_alconna(
aliases={"更新轻雪"},
command=Alconna(
"update-liteyuki"
),
permission=SUPERUSER
)
reload_liteyuki = on_alconna(
aliases={"重启轻雪"},
command=Alconna(
"reload-liteyuki"
),
permission=SUPERUSER
)
reload_resources = on_alconna(
aliases={"重载资源"},
command=Alconna(
"reload-resources"
),
permission=SUPERUSER
)
cmd_config = on_alconna(
aliases={"配置"},
command=Alconna(
"config",
Subcommand(
"set",
Args["key", str]["value", Any],
alias=["设置"],
),
Subcommand(
"get",
Args["key", str, None],
alias=["查询"]
)
),
permission=SUPERUSER
)
switch_image_mode = on_alconna(
aliases={"切换图片模式"},
command=Alconna(
"switch-image-mode"
),
permission=SUPERUSER
)
@liteyuki.handle()
async def _(bot: T_Bot):
await liteyuki.finish(f"Hello, Liteyuki!\nBot {bot.self_id}")
@update_liteyuki.handle()
).handle()
async def _(bot: T_Bot, event: T_MessageEvent):
# 使用git pull更新
ulang = get_user_lang(str(event.user_id))
@ -117,20 +70,43 @@ async def _(bot: T_Bot, event: T_MessageEvent):
nonebot.logger.error(f"Pull from {origin} failed: {e}")
reply = "Liteyuki updated!\n"
reply += f"```\n{logs}\n```\n"
btn_restart = md.cmd(ulang.get("liteyuki.restart_now"), "reload-liteyuki")
btn_restart = md.btn_cmd(ulang.get("liteyuki.restart_now"), "reload-liteyuki")
pip.main(["install", "-r", "requirements.txt"])
reply += f"{ulang.get('liteyuki.update_restart', RESTART=btn_restart)}"
await md.send_md(reply, bot, event=event, at_sender=False)
@reload_liteyuki.handle()
async def _():
await reload_liteyuki.send("Liteyuki reloading")
@on_alconna(
aliases={"重启轻雪"},
command=Alconna(
"reload-liteyuki"
),
permission=SUPERUSER
).handle()
async def _(matcher: Matcher):
await matcher.send("Liteyuki reloading")
Reloader.reload(3)
@cmd_config.handle()
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
@on_alconna(
aliases={"配置"},
command=Alconna(
"config",
Subcommand(
"set",
Args["key", str]["value", Any],
alias=["设置"],
),
Subcommand(
"get",
Args["key", str, None],
alias=["查询"]
)
),
permission=SUPERUSER
).handle()
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, matcher: Matcher):
ulang = get_user_lang(str(event.user_id))
stored_config: StoredConfig = common_db.first(StoredConfig(), default=StoredConfig())
if result.subcommands.get("set"):
@ -141,7 +117,7 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
pass
stored_config.config[key] = value
common_db.upsert(stored_config)
await cmd_config.finish(f"{ulang.get('liteyuki.config_set_success', KEY=key, VAL=value)}")
await matcher.finish(f"{ulang.get('liteyuki.config_set_success', KEY=key, VAL=value)}")
elif result.subcommands.get("get"):
key = result.subcommands.get("get").args.get("key")
file_config = load_from_yaml("config.yml")
@ -162,19 +138,14 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
await md.send_md(reply, bot, event=event)
@reload_resources.handle()
async def _(event: T_MessageEvent):
ulang = get_user_lang(str(event.user_id))
load_resources()
await reload_resources.finish(
ulang.get("liteyuki.reload_resources_success",
NUM=len(get_loaded_resource_packs())
)
)
@switch_image_mode.handle()
async def _(bot: T_Bot, event: T_MessageEvent):
@on_alconna(
aliases={"切换图片模式"},
command=Alconna(
"switch-image-mode"
),
permission=SUPERUSER
).handle()
async def _(event: T_MessageEvent, matcher: Matcher):
global markdown_image
# 切换图片模式False以图片形式发送True以markdown形式发送
ulang = get_user_lang(str(event.user_id))
@ -182,12 +153,20 @@ async def _(bot: T_Bot, event: T_MessageEvent):
stored_config.config["markdown_image"] = not stored_config.config.get("markdown_image", False)
markdown_image = stored_config.config["markdown_image"]
common_db.upsert(stored_config)
await switch_image_mode.finish(ulang.get("liteyuki.image_mode_on" if stored_config.config["markdown_image"] else "liteyuki.image_mode_off"))
await matcher.finish(ulang.get("liteyuki.image_mode_on" if stored_config.config["markdown_image"] else "liteyuki.image_mode_off"))
@on_alconna(
command=Alconna(
"liteyuki-docs",
),
aliases={"轻雪文档"},
).handle()
async def _(matcher: Matcher):
matcher.finish("https://bot.liteyuki.icu/usage")
# system hook
@Bot.on_calling_api
@Bot.on_calling_api # 图片模式检测
async def test_for_md_image(bot: T_Bot, api: str, data: dict):
if api in ["send_msg", "send_private_msg", "send_group_msg"] and markdown_image and data.get("user_id") != bot.self_id:
if api == "send_msg" and data.get("message_type") == "private" or api == "send_private_msg":

View File

@ -37,7 +37,8 @@ async def _(bot: T_Bot, event: T_MessageEvent):
{
"data": await get_stats_data(bot.self_id, ulang.lang_code)
},
debug=True
debug=True,
wait=1
)
await stats.finish(MessageSegment.image(image))
@ -233,10 +234,4 @@ async def get_stats_data(self_id: str = None, lang: str = None) -> dict:
"total_trans": ulang.get("main.monitor.total"),
}
# for ps_name, ps_mem in process_mem.items():
# templ["memTags"].insert(
# 0,
# f"{ps_name} {convert_size(ps_mem, 1)}"
# )
return templ

View File

@ -157,14 +157,14 @@ class Minesweeper:
print([d.value for d in row])
for dot in row:
if dot.mask and not dot.flagged:
text += md.cmd(self.MASK, f"minesweeper reveal {dot.row} {dot.col}")
text += md.btn_cmd(self.MASK, f"minesweeper reveal {dot.row} {dot.col}")
elif dot.flagged:
text += md.cmd(self.FLAG, f"minesweeper mark {dot.row} {dot.col}")
text += md.btn_cmd(self.FLAG, f"minesweeper mark {dot.row} {dot.col}")
else:
text += self.NUMS[dot.value]
text += dis
text += "\n"
btn_mark = md.cmd("标记", f"minesweeper mark ", enter=False)
btn_end = md.cmd("结束", "minesweeper end", enter=True)
btn_mark = md.btn_cmd("标记", f"minesweeper mark ", enter=False)
btn_end = md.btn_cmd("结束", "minesweeper end", enter=True)
text += f" {btn_mark} {btn_end}"
return text

View File

@ -16,9 +16,9 @@ from .common import *
require("nonebot_plugin_alconna")
from nonebot_plugin_alconna import Alconna, Args, Subcommand, on_alconna
npm_alc = on_alconna(
nps = on_alconna(
Alconna(
"npm",
"nps",
Subcommand(
"update",
alias=["u"],
@ -36,19 +36,15 @@ npm_alc = on_alconna(
Subcommand(
"uninstall",
Args["plugin_name", str],
alias=["rm", "移除", "卸载"],
),
Subcommand(
"list",
alias=["l", "ls", "列表"],
alias=["r", "rm", "卸载"],
)
),
aliases={"插件"},
aliases={"插件商店"},
permission=SUPERUSER,
)
@npm_alc.handle()
@nps.handle()
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
ulang = get_user_lang(str(event.user_id))
@ -58,9 +54,9 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
if result.subcommands.get("update"):
r = await npm_update()
if r:
await npm_alc.finish(ulang.get("npm.store_update_success"))
await nps.finish(ulang.get("npm.store_update_success"))
else:
await npm_alc.finish(ulang.get("npm.store_update_failed"))
await nps.finish(ulang.get("npm.store_update_failed"))
elif result.subcommands.get("search"):
keywords: list[str] = result.subcommands["search"].args.get("keywords")
@ -71,9 +67,9 @@ 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.cmd(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)
btn_install = md.btn_cmd(ulang.get("npm.install"), "npm install %s" % plugin.module_name)
link_page = md.btn_link(ulang.get("npm.homepage"), plugin.homepage)
link_pypi = md.btn_link(ulang.get("npm.pypi"), plugin.homepage)
reply += (f"\n# **{plugin.name}**\n"
f"\n> **{plugin.desc}**\n"
@ -89,14 +85,14 @@ 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))
await nps.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))
await nps.finish(ulang.get("npm.plugin_not_found", NAME=plugin_module_name))
homepage_btn = md.cmd(ulang.get("npm.homepage"), store_plugin.homepage)
homepage_btn = md.btn_cmd(ulang.get("npm.homepage"), store_plugin.homepage)
if r:
r_load = nonebot.load_plugin(plugin_module_name) # 加载插件
@ -114,7 +110,7 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
event=event
)
else:
await npm_alc.finish(ulang.get("npm.plugin_already_installed", NAME=store_plugin.name))
await nps.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 md.send_md(
@ -138,9 +134,9 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
if found_installed_plugin:
plugin_db.delete(InstalledPlugin(), "module_name = ?", plugin_module_name)
reply = f"{ulang.get('npm.uninstall_success', NAME=found_installed_plugin.module_name)}"
await npm_alc.finish(reply)
await nps.finish(reply)
else:
await npm_alc.finish(ulang.get("npm.plugin_not_installed", NAME=plugin_module_name))
await nps.finish(ulang.get("npm.plugin_not_installed", NAME=plugin_module_name))
async def npm_update() -> bool:

View File

@ -19,7 +19,7 @@ from .installer import get_store_plugin, npm_update
from ...utils.tools import clamp
require("nonebot_plugin_alconna")
from nonebot_plugin_alconna import on_alconna, Alconna, Args, Arparma
from nonebot_plugin_alconna import on_alconna, Alconna, Args, Arparma, Subcommand
list_plugins = on_alconna(
Alconna(
@ -29,27 +29,40 @@ list_plugins = on_alconna(
aliases={"列出插件", "插件列表"}
)
toggle_plugin = on_alconna(
Alconna(
"enable",
Args["plugin_name", str],
),
aliases={"disable", "启用", "停用"}
)
npm = on_alconna(
aliases={"插件管理"},
command=Alconna(
"npm",
# Args["plugin_name", str],
Subcommand(
"enable",
Args["plugin_name", str],
alias=["启用"],
toggle_plugin_global = on_alconna(
Alconna(
"enable-global",
Args["plugin_name", str],
),
Subcommand(
"disable",
Args["plugin_name", str],
alias=["停用"],
),
Subcommand(
"global-enable",
Args["plugin_name", str],
alias=["全局启用"],
),
Subcommand(
"global-disable",
Args["plugin_name", str],
alias=["全局停用"],
),
),
permission=SUPERUSER,
aliases={"disable-global", "全局启用", "全局停用"}
)
@list_plugins.handle()
async def _(event: T_MessageEvent, bot: T_Bot, result: Arparma):
lang = get_user_lang(str(event.user_id))
ulang = get_user_lang(str(event.user_id))
if not os.path.exists("data/liteyuki/plugins.json"):
await npm_update()
@ -60,31 +73,29 @@ async def _(event: T_MessageEvent, bot: T_Bot, result: Arparma):
page = clamp(result.args.get("page"), 1, total)
# 已加载插件 | 总计10 | 第1/3页
reply = (f"# {lang.get('npm.loaded_plugins')} | "
f"{lang.get('npm.total', TOTAL=len(nonebot.get_loaded_plugins()))} | "
f"{lang.get('npm.page', PAGE=page, TOTAL=total)} \n***\n")
reply = (f"# {ulang.get('npm.loaded_plugins')} | "
f"{ulang.get('npm.total', TOTAL=len(nonebot.get_loaded_plugins()))} | "
f"{ulang.get('npm.page', PAGE=page, TOTAL=total)} \n***\n")
for plugin in loaded_plugin_list[(page - 1) * num_per_page: min(page * num_per_page, len(loaded_plugin_list))]:
# 检查是否有 metadata 属性
# 添加帮助按钮
btn_usage = md.cmd(lang.get("npm.usage"), f"help {plugin.module_name}", False)
btn_usage = md.btn_cmd(ulang.get("npm.usage"), f"help {plugin.module_name}", False)
store_plugin = await get_store_plugin(plugin.module_name)
session_enable = get_plugin_session_enable(event, plugin.module_name)
if store_plugin:
btn_homepage = md.link(lang.get("npm.homepage"), store_plugin.homepage)
btn_homepage = md.btn_link(ulang.get("npm.homepage"), store_plugin.homepage)
show_name = store_plugin.name
elif plugin.metadata:
if plugin.metadata.extra.get("liteyuki"):
btn_homepage = md.link(lang.get("npm.homepage"), "https://github.com/snowykami/LiteyukiBot")
btn_homepage = md.btn_link(ulang.get("npm.homepage"), "https://github.com/snowykami/LiteyukiBot")
else:
btn_homepage = lang.get("npm.homepage")
btn_homepage = ulang.get("npm.homepage")
show_name = plugin.metadata.name
else:
btn_homepage = lang.get("npm.homepage")
btn_homepage = ulang.get("npm.homepage")
show_name = plugin.name
lang.get("npm.no_description")
ulang.get("npm.no_description")
if plugin.metadata:
reply += f"\n**{md.escape(show_name)}**\n"
@ -95,11 +106,10 @@ async def _(event: T_MessageEvent, bot: T_Bot, result: Arparma):
if await GROUP_ADMIN(bot, event) or await GROUP_OWNER(bot, event) or await SUPERUSER(bot, event):
# 添加启用/停用插件按钮
cmd_toggle = f"{'disable' if session_enable else 'enable'} {plugin.module_name}"
text_toggle = lang.get("npm.disable" if session_enable else "npm.enable")
cmd_toggle = f"npm {'disable' if session_enable else 'enable'} {plugin.module_name}"
text_toggle = ulang.get("npm.disable" if session_enable else "npm.enable")
can_be_toggle = get_plugin_can_be_toggle(plugin.module_name)
btn_toggle = text_toggle if not can_be_toggle else md.cmd(text_toggle, cmd_toggle)
btn_toggle = text_toggle if not can_be_toggle else md.btn_cmd(text_toggle, cmd_toggle)
reply += f" {btn_toggle}"
if await SUPERUSER(bot, event):
@ -107,20 +117,18 @@ async def _(event: T_MessageEvent, bot: T_Bot, result: Arparma):
# 添加移除插件和全局切换按钮
global_enable = get_plugin_global_enable(plugin.module_name)
btn_uninstall = (
md.cmd(lang.get("npm.uninstall"), f'npm uninstall {plugin.module_name}')) if plugin_in_database else lang.get(
md.btn_cmd(ulang.get("npm.uninstall"), f'npm uninstall {plugin.module_name}')) if plugin_in_database else ulang.get(
'npm.uninstall')
btn_toggle_global_text = lang.get("npm.disable_global" if global_enable else "npm.enable_global")
cmd_toggle_global = f"{'disable-global' if global_enable else 'enable-global'} {plugin.module_name}"
btn_toggle_global = btn_toggle_global_text if not can_be_toggle else md.cmd(btn_toggle_global_text, cmd_toggle_global)
btn_toggle_global_text = ulang.get("npm.disable_global" if global_enable else "npm.enable_global")
cmd_toggle_global = f"npm {'global-disable' if global_enable else 'global-enable'} {plugin.module_name}"
btn_toggle_global = btn_toggle_global_text if not can_be_toggle else md.btn_cmd(btn_toggle_global_text, cmd_toggle_global)
reply += f" {btn_uninstall} {btn_toggle_global}"
reply += "\n\n***\n"
await md.send_md(reply, bot, event=event)
@toggle_plugin.handle()
@npm.handle()
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
if not os.path.exists("data/liteyuki/plugins.json"):
await npm_update()
@ -128,103 +136,96 @@ 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")
# 支持对自定义command_start的判断
toggle = result.header_result in [prefix + header for prefix in bot.config.command_start for header in ["enable-plugin", "启用"]] # 判断是启用还是停用
if result.subcommands.get("enable") or result.subcommands.get("disable"):
session_enable = get_plugin_session_enable(event, plugin_module_name) # 获取插件当前状态
toggle = result.subcommands.get("enable") is not None
default_enable = get_plugin_default_enable(plugin_module_name) # 获取插件默认状态
session_enable = get_plugin_session_enable(event, plugin_module_name) # 获取插件当前状态
can_be_toggled = get_plugin_can_be_toggle(plugin_module_name) # 获取插件是否可以被启用/停用
default_enable = get_plugin_default_enable(plugin_module_name) # 获取插件默认状态
if not can_be_toggled:
await toggle_plugin.finish(ulang.get("npm.plugin_cannot_be_toggled", NAME=plugin_module_name))
can_be_toggled = get_plugin_can_be_toggle(plugin_module_name) # 获取插件是否可以被启用/停用
if session_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 not can_be_toggled:
await npm.finish(ulang.get("npm.plugin_cannot_be_toggled", NAME=plugin_module_name))
if session_enable == toggle:
await npm.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(Group(), "group_id = ?", event.group_id, default=Group(group_id=str(event.group_id)))
else:
raise FinishedException(ulang.get("Permission Denied"))
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)
if event.message_type == "private":
user_db.upsert(session)
session = user_db.first(User(), "user_id = ?", event.user_id, default=User(user_id=event.user_id))
else:
group_db.upsert(session)
except Exception as e:
print(e)
await toggle_plugin.finish(
if await GROUP_ADMIN(bot, event) or await GROUP_OWNER(bot, event) or await SUPERUSER(bot, event):
session = group_db.first(Group(), "group_id = ?", event.group_id, default=Group(group_id=str(event.group_id)))
else:
raise FinishedException(ulang.get("Permission Denied"))
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)
if event.message_type == "private":
user_db.upsert(session)
else:
group_db.upsert(session)
except Exception as e:
print(e)
await npm.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))
)
await npm.finish(
ulang.get(
"npm.toggle_failed",
"npm.toggle_success",
NAME=plugin_module_name,
STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"),
ERROR=str(e))
STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"))
)
elif result.subcommands.get("global-enable") or result.subcommands.get("global-disable") and await SUPERUSER(bot, event):
toggle = result.subcommands.get("global-enable") is not None
can_be_toggled = get_plugin_can_be_toggle(plugin_module_name)
if not can_be_toggled:
await npm.finish(ulang.get("npm.plugin_cannot_be_toggled", NAME=plugin_module_name))
await toggle_plugin.finish(
ulang.get(
"npm.toggle_success",
NAME=plugin_module_name,
STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"))
)
global_enable = get_plugin_global_enable(plugin_module_name)
if global_enable == toggle:
await npm.finish(
ulang.get("npm.plugin_already", NAME=plugin_module_name, STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable")))
try:
plugin = plugin_db.first(GlobalPlugin(), "module_name = ?", plugin_module_name, default=GlobalPlugin(module_name=plugin_module_name))
if toggle:
plugin.enabled = True
else:
plugin.enabled = False
plugin_db.upsert(plugin)
except Exception as e:
print(e)
await npm.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))
)
@toggle_plugin_global.handle()
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
if not os.path.exists("data/liteyuki/plugins.json"):
await npm_update()
# 判断会话类型
ulang = get_user_lang(str(event.user_id))
plugin_module_name = result.args.get("plugin_name")
# 支持对自定义command_start的判断
toggle = result.header_result in [prefix + header for prefix in bot.config.command_start for header in ["enable-global", "全局启用"]]
can_be_toggled = get_plugin_can_be_toggle(plugin_module_name)
if not can_be_toggled:
await toggle_plugin_global.finish(ulang.get("npm.plugin_cannot_be_toggled", NAME=plugin_module_name))
global_enable = get_plugin_global_enable(plugin_module_name)
if global_enable == toggle:
await toggle_plugin_global.finish(
ulang.get("npm.plugin_already", NAME=plugin_module_name, STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable")))
try:
plugin = plugin_db.first(GlobalPlugin(), "module_name = ?", plugin_module_name, default=GlobalPlugin(module_name=plugin_module_name))
if toggle:
plugin.enabled = True
else:
plugin.enabled = False
plugin_db.upsert(plugin)
except Exception as e:
print(e)
await toggle_plugin_global.finish(
await npm.finish(
ulang.get(
"npm.toggle_failed",
"npm.toggle_success",
NAME=plugin_module_name,
STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"),
ERROR=str(e))
STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"))
)
await toggle_plugin_global.finish(
ulang.get(
"npm.toggle_success",
NAME=plugin_module_name,
STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"))
)
@run_preprocessor
async def pre_handle(event: Event, matcher: Matcher):

View File

@ -1,18 +1,173 @@
# 轻雪资源包管理器
import os
import yaml
from nonebot import require
from nonebot.permission import SUPERUSER
from nonebot_plugin_alconna import on_alconna, Alconna, Args, Subcommand, Arparma
from liteyuki.utils.ly_typing import T_Bot
from liteyuki.utils.language import get_user_lang
from liteyuki.utils.ly_typing import T_Bot, T_MessageEvent
from liteyuki.utils.message import Markdown as md
from liteyuki.utils.resource import (ResourceMetadata, add_resource_pack, change_priority, check_exist, check_status, get_loaded_resource_packs, get_resource_metadata, load_resources, remove_resource_pack)
list_rp = on_alconna(
aliases={"列出资源包", "资源包列表"},
require("nonebot_plugin_alconna")
from nonebot_plugin_alconna import Alconna, Args, on_alconna, Arparma, Subcommand
rpm = on_alconna(
aliases={"资源包"},
command=Alconna(
"list",
Args["page", int, 1]["num", int, 10],
"rpm",
Subcommand(
"list",
Args["page", int, 1]["num", int, 10],
alias=["ls", "列表", "列出"],
),
Subcommand(
"load",
Args["name", str],
alias=["安装"],
),
Subcommand(
"unload",
Args["name", str],
alias=["卸载"],
),
Subcommand(
"up",
Args["name", str],
alias=["上移"],
),
Subcommand(
"down",
Args["name", str],
alias=["下移"],
),
Subcommand(
"top",
Args["name", str],
alias=["置顶"],
),
Subcommand(
"reload",
alias=["重载"],
),
),
permission=SUPERUSER
)
@list_rp.handle()
async def _(bot: T_Bot):
pass
@rpm.handle()
async def _(bot: T_Bot, event: T_MessageEvent, result: Arparma):
ulang = get_user_lang(str(event.user_id))
reply = ""
if result.subcommands.get("list"):
loaded_rps = get_loaded_resource_packs()
reply += f"{ulang.get('liteyuki.loaded_resources', NUM=len(loaded_rps))}\n"
for rp in loaded_rps:
btn_unload = md.btn_cmd(
ulang.get("npm.uninstall"),
f"rpm unload {rp.folder}"
)
btn_move_up = md.btn_cmd(
ulang.get("rpm.move_up"),
f"rpm up {rp.folder}"
)
btn_move_down = md.btn_cmd(
ulang.get("rpm.move_down"),
f"rpm down {rp.folder}"
)
btn_move_top = md.btn_cmd(
ulang.get("rpm.move_top"),
f"rpm top {rp.folder}"
)
# 添加新行
reply += (f"\n**{md.escape(rp.name)}**({md.escape(rp.folder)})\n\n"
f"> {btn_move_up} {btn_move_down} {btn_move_top} {btn_unload}\n\n***")
reply += f"\n\n{ulang.get('liteyuki.unloaded_resources')}\n"
loaded_folders = [rp.folder for rp in get_loaded_resource_packs()]
for folder in os.listdir("resources"):
if folder not in loaded_folders and os.path.exists(os.path.join("resources", folder, "metadata.yml")):
metadata = ResourceMetadata(
**yaml.load(
open(
os.path.join("resources", folder, "metadata.yml"),
encoding="utf-8"
),
Loader=yaml.FullLoader
)
)
metadata.folder = folder
metadata.path = os.path.join("resources", folder)
btn_load = md.btn_cmd(
ulang.get("npm.install"),
f"rpm load {metadata.folder}"
)
# 添加新行
reply += (f"\n**{md.escape(metadata.name)}**({md.escape(metadata.folder)})\n\n"
f"> {btn_load}\n\n***")
elif result.subcommands.get("load") or result.subcommands.get("unload"):
load = result.subcommands.get("load") is not None
rp_name = result.args.get("name")
r = False # 操作结果
if check_exist(rp_name):
if load != check_status(rp_name):
# 状态不同
if load:
r = add_resource_pack(rp_name)
else:
r = remove_resource_pack(rp_name)
rp_meta = get_resource_metadata(rp_name)
reply += ulang.get(
f"liteyuki.{'load' if load else 'unload'}_resource_{'success' if r else 'failed'}",
NAME=rp_meta.name
)
else:
# 重复操作
reply += ulang.get(f"liteyuki.resource_already_{'load' if load else 'unload'}ed", NAME=rp_name)
else:
reply += ulang.get("liteyuki.resource_not_found", NAME=rp_name)
if r:
btn_reload = md.btn_cmd(
ulang.get("liteyuki.reload_resources"),
f"rpm reload"
)
reply += "\n" + ulang.get("liteyuki.need_reload", BTN=btn_reload)
elif result.subcommands.get("up") or result.subcommands.get("down") or result.subcommands.get("top"):
rp_name = result.args.get("name")
if result.subcommands.get("up"):
delta = -1
elif result.subcommands.get("down"):
delta = 1
else:
delta = 0
if check_exist(rp_name):
if check_status(rp_name):
r = change_priority(rp_name, delta)
reply += ulang.get(f"liteyuki.change_priority_{'success' if r else 'failed'}", NAME=rp_name)
if r:
btn_reload = md.btn_cmd(
ulang.get("liteyuki.reload_resources"),
f"rpm reload"
)
reply += "\n" + ulang.get("liteyuki.need_reload", BTN=btn_reload)
else:
reply += ulang.get("liteyuki.resource_not_found", NAME=rp_name)
else:
reply += ulang.get("liteyuki.resource_not_found", NAME=rp_name)
elif result.subcommands.get("reload"):
load_resources()
reply = ulang.get(
"liteyuki.reload_resources_success",
NUM=len(get_loaded_resource_packs())
)
else:
btn_reload = md.btn_cmd(
ulang.get("liteyuki.reload_resources"),
f"rpm reload"
)
btn_list = md.btn_cmd(
ulang.get("liteyuki.list_resources"),
f"rpm list"
)
reply += f"{btn_list} {btn_reload}"
await md.send_md(reply, bot, event=event)

View File

@ -90,8 +90,8 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
continue
val = profile.dict()[key]
key_text = ulang.get(f"user.profile.{key}")
btn_set = md.cmd(ulang.get("user.profile.edit"), f"profile set {key}",
enter=True if key in enter_attr else False)
btn_set = md.btn_cmd(ulang.get("user.profile.edit"), f"profile set {key}",
enter=True if key in enter_attr else False)
reply += (f"\n**{key_text}** **{val}**\n"
f"\n> {ulang.get(f'user.profile.{key}.desc')}"
f"\n> {btn_set} \n\n***\n")
@ -117,11 +117,11 @@ def get_profile_menu(key: str, ulang: Language) -> Optional[str]:
reply = f"**{setting_name} {ulang.get('user.profile.settings')}**\n***\n"
if key == "lang":
for lang_code, lang_name in get_all_lang().items():
btn_set = md.cmd(ulang.get("user.profile.set"), f"profile set {key} {lang_code}")
btn_set = md.btn_cmd(ulang.get("user.profile.set"), f"profile set {key} {lang_code}")
reply += f"\n{btn_set} | **{lang_name}** - {lang_code}\n***\n"
elif key == "timezone":
for tz in representative_timezones_list:
btn_set_tz = md.cmd(tz, f"profile set {key} {tz}")
btn_set_tz = md.btn_cmd(tz, f"profile set {key} {tz}")
reply += f"{btn_set_tz}\n"
return reply

View File

@ -100,4 +100,23 @@ liteyuki.image_mode_off=关闭Markdown图片模式
npm.page=第{PAGE}/{TOTAL}页
main.monitor.free=空闲
liteyuki.reload_resources_success=资源重载成功,共计{NUM}个资源包
liteyuki.invalid_command=无效的命令或参数
liteyuki.reload_resources=重载资源
liteyuki.list_resources=资源包列表
liteyuki.reload_resources_success=资源重载成功,共计 {NUM} 个资源包
liteyuki.loaded_resources=已加载 {NUM} 个资源包,按照优先级排序
liteyuki.unloaded_resources=未加载资源包
liteyuki.load_resource_success=资源包 {NAME} 加载成功
liteyuki.unload_resource_success=资源包 {NAME} 卸载成功
liteyuki.load_resource_failed=资源包 {NAME} 加载失败
liteyuki.unload_resource_failed=资源包 {NAME} 卸载失败
liteyuki.resource_not_found=资源包 {NAME} 不存在或不可操作
liteyuki.resource_already_loaded=资源包 {NAME} 已加载,请勿重复操作
liteyuki.resource_already_unloaded=资源包 {NAME} 已卸载,请勿重复操作
liteyuki.need_reload=请{BTN}重载以应用这些更新
liteyuki.dont_repeat=请勿重复操作
liteyuki.change_priority_success=资源包 {NAME} 优先级修改成功
liteyuki.change_priority_failed=资源包 {NAME} 优先级修改失败
rpm.move_up=上移
rpm.move_down=下移
rpm.move_top=置顶

View File

@ -94,7 +94,7 @@
}
</style>
<script type="text/javascript" src="https://cdn.jsdelivr.net/npm/echarts@4.3.0/dist/echarts.min.js"></script>
<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/echarts/4.3.0/echarts.min.js"></script>
</head>
<body>

View File

@ -6,7 +6,7 @@ import sys
import nonebot
__NAME__ = "LiteyukiBot"
__VERSION__ = "6.2.6" # 60201
__VERSION__ = "6.2.7" # 60201
import requests

View File

@ -10,27 +10,6 @@ require("nonebot_plugin_htmlrender")
from nonebot_plugin_htmlrender import *
# async def html2image(
# html: str,
# wait: int = 0,
# template_path: str = None,
# scale_factor: float = 2,
# **kwargs
# ) -> bytes:
# """
# Args:
# html: str: HTML 正文
# wait: 等待时间
# template_path: 模板路径
# scale_factor: 缩放因子,越高越清晰
# **kwargs: page 参数
#
# Returns:
#
# """
# return await html_to_pic(html, wait=wait, template_path=template_path, scale_factor=scale_factor)
async def template2html(
template: str,
templates: dict,
@ -77,6 +56,7 @@ async def template2image(
}
template_path = os.path.dirname(template)
template_name = os.path.basename(template)
print(template_path, template_name)
if debug:
raw_html = await template_to_html(

View File

@ -12,7 +12,6 @@ import nonebot
from .config import config
from .data_manager import User, user_db
_default_lang_code = "en"
_language_data = {
"en": {
"name": "English",
@ -102,7 +101,7 @@ def load_from_dict(data: dict, lang_code: str):
class Language:
def __init__(self, lang_code: str = None, fallback_lang_code: str = "en"):
def __init__(self, lang_code: str = None, fallback_lang_code: str = "zh-CN"):
if lang_code is None:
lang_code = config.get("default_language", get_default_lang())
self.lang_code = lang_code

View File

@ -185,7 +185,7 @@ class Markdown:
# 等林文轩修好Lagrange.OneBot再说
@staticmethod
def cmd(name: str, cmd: str, reply: bool = False, enter: bool = True) -> str:
def btn_cmd(name: str, cmd: str, reply: bool = False, enter: bool = True) -> str:
"""生成点击回调按钮
Args:
name: 按钮显示内容
@ -202,7 +202,7 @@ class Markdown:
return f"[{name}](mqqapi://aio/inlinecmd?command={quote(cmd)}&reply={str(reply).lower()}&enter={str(enter).lower()})"
@staticmethod
def link(name: str, url: str) -> str:
def btn_link(name: str, url: str) -> str:
"""生成点击链接按钮
Args:
name: 链接显示内容

View File

@ -1,21 +1,25 @@
import json
import os
import shutil
from typing import Any
import nonebot
import yaml
from typing import Any
from liteyuki.utils.data import LiteModel
from .data import LiteModel
from .language import get_default_lang
_loaded_resource_packs: list["ResourceMetadata"] = [] # 按照加载顺序排序
temp_resource_root = "data/liteyuki/resources"
lang = get_default_lang()
class ResourceMetadata(LiteModel):
name: str = "Unknown"
version: str = "0.0.1"
description: str = "Unknown"
path: str
path: str = ""
folder: str = ""
def load_resource_from_dir(path: str):
@ -36,10 +40,11 @@ def load_resource_from_dir(path: str):
relative_path = os.path.relpath(os.path.join(root, file), path)
copy_file(os.path.join(root, file), os.path.join(temp_resource_root, relative_path))
metadata["path"] = path
metadata["folder"] = os.path.basename(path)
if os.path.exists(os.path.join(path, "lang")):
from liteyuki.utils.language import load_from_dir
load_from_dir(os.path.join(path, "lang"))
_loaded_resource_packs.append(ResourceMetadata(**metadata))
_loaded_resource_packs.insert(0, ResourceMetadata(**metadata))
def get_path(path: str, abs_path: bool = False, default: Any = None) -> str | Any:
@ -76,7 +81,7 @@ def get_files(path: str, abs_path: bool = False) -> list[str]:
def get_loaded_resource_packs() -> list[ResourceMetadata]:
"""
获取已加载的资源包
获取已加载的资源包优先级从前到后
Returns: 资源包列表
"""
return _loaded_resource_packs
@ -106,6 +111,117 @@ def load_resources():
standard_resource_path = "liteyuki/resources"
load_resource_from_dir(standard_resource_path)
# 加载其他资源包
if os.path.exists("resources"):
for resource in os.listdir("resources"):
load_resource_from_dir(os.path.join("resources", resource))
if not os.path.exists("resources"):
os.makedirs("resources", exist_ok=True)
if not os.path.exists("resources/index.json"):
json.dump([], open("resources/index.json", "w", encoding="utf-8"))
resource_index: list[str] = json.load(open("resources/index.json", "r", encoding="utf-8"))
resource_index.reverse() # 优先级高的后加载,但是排在前面
for resource in resource_index:
load_resource_from_dir(os.path.join("resources", resource))
def check_status(name: str) -> bool:
"""
检查资源包是否已加载
Args:
name: 资源包名称文件夹名
Returns: 是否已加载
"""
return name in [rp.folder for rp in get_loaded_resource_packs()]
def check_exist(name: str) -> bool:
"""
检查资源包文件夹是否存在于resources文件夹
Args:
name: 资源包名称文件夹名
Returns: 是否存在
"""
return os.path.exists(os.path.join("resources", name, "metadata.yml"))
def add_resource_pack(name: str) -> bool:
"""
添加资源包该操作仅修改index.json文件不会加载资源包要生效请重载资源
Args:
name: 资源包名称文件夹名
Returns:
"""
if check_exist(name):
old_index: list[str] = json.load(open("resources/index.json", "r", encoding="utf-8"))
if name not in old_index:
old_index.append(name)
json.dump(old_index, open("resources/index.json", "w", encoding="utf-8"))
load_resource_from_dir(os.path.join("resources", name))
return True
else:
nonebot.logger.warning(lang.get("liteyuki.resource_loaded", name=name))
return False
else:
nonebot.logger.warning(lang.get("liteyuki.resource_not_exist", name=name))
return False
def remove_resource_pack(name: str) -> bool:
"""
移除资源包该操作仅修改加载索引要生效请重载资源
Args:
name: 资源包名称文件夹名
Returns:
"""
if check_exist(name):
old_index: list[str] = json.load(open("resources/index.json", "r", encoding="utf-8"))
if name in old_index:
old_index.remove(name)
json.dump(old_index, open("resources/index.json", "w", encoding="utf-8"))
return True
else:
nonebot.logger.warning(lang.get("liteyuki.resource_not_loaded", name=name))
return False
else:
nonebot.logger.warning(lang.get("liteyuki.resource_not_exist", name=name))
return False
def change_priority(name: str, delta: int) -> bool:
"""
修改资源包优先级
Args:
name: 资源包名称文件夹名
delta: 优先级变化正数表示后移负数表示前移0表示移到最前
Returns:
"""
# 正数表示前移,负数表示后移
old_resource_list: list[str] = json.load(open("resources/index.json", "r", encoding="utf-8"))
new_resource_list = old_resource_list.copy()
if name in old_resource_list:
index = old_resource_list.index(name)
if 0 <= index + delta < len(old_resource_list):
new_index = index + delta
new_resource_list.remove(name)
new_resource_list.insert(new_index, name)
json.dump(new_resource_list, open("resources/index.json", "w", encoding="utf-8"))
return True
else:
nonebot.logger.warning("Priority change failed, out of range")
return False
else:
nonebot.logger.debug("Priority change failed, resource not loaded")
return False
def get_resource_metadata(name: str) -> ResourceMetadata:
"""
获取资源包元数据
Args:
name: 资源包名称文件夹名
Returns:
"""
for rp in get_loaded_resource_packs():
if rp.folder == name:
return rp
return ResourceMetadata()