mirror of
https://github.com/LiteyukiStudio/LiteyukiBot.git
synced 2025-09-14 07:36:24 +00:00
🐛 fix 通道类回调函数在进程间传递时无法序列号的问题
This commit is contained in:
22
src/nonebot_plugins/liteyuki_pacman/__init__.py
Normal file
22
src/nonebot_plugins/liteyuki_pacman/__init__.py
Normal file
@ -0,0 +1,22 @@
|
||||
from nonebot.plugin import PluginMetadata
|
||||
from .npm import *
|
||||
from .rpm import *
|
||||
|
||||
__author__ = "snowykami"
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="轻雪包管理器",
|
||||
description="本地插件管理和插件商店支持，资源包管理，支持启用/停用，安装/卸载插件",
|
||||
usage=(
|
||||
"npm list\n"
|
||||
"npm enable/disable <plugin_name>\n"
|
||||
"npm search <keywords...>\n"
|
||||
"npm install/uninstall <plugin_name>\n"
|
||||
),
|
||||
type="application",
|
||||
homepage="https://github.com/snowykami/LiteyukiBot",
|
||||
extra={
|
||||
"liteyuki": True,
|
||||
"toggleable" : False,
|
||||
"default_enable" : True,
|
||||
}
|
||||
)
|
||||
255
src/nonebot_plugins/liteyuki_pacman/common.py
Normal file
255
src/nonebot_plugins/liteyuki_pacman/common.py
Normal file
@ -0,0 +1,255 @@
|
||||
import json
|
||||
from typing import Optional
|
||||
|
||||
import aiofiles
|
||||
import nonebot.plugin
|
||||
from nonebot.adapters import satori
|
||||
|
||||
from src.utils import event as event_utils
|
||||
from src.utils.base.data import LiteModel
|
||||
from src.utils.base.data_manager import GlobalPlugin, Group, User, group_db, plugin_db, user_db
|
||||
from src.utils.base.ly_typing import T_MessageEvent
|
||||
|
||||
__group_data = {} # 群数据缓存, {group_id: Group}
|
||||
__user_data = {} # 用户数据缓存, {user_id: User}
|
||||
__default_enable = {} # 插件默认启用状态缓存, {plugin_name: bool} static
|
||||
__global_enable = {} # 插件全局启用状态缓存, {plugin_name: bool} dynamic
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
def get_plugin_exist(plugin_name: str) -> bool:
|
||||
"""
|
||||
获取插件是否存在于加载列表
|
||||
Args:
|
||||
plugin_name:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
for plugin in nonebot.plugin.get_loaded_plugins():
|
||||
if plugin.name == plugin_name:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
async def get_store_plugin(plugin_name: str) -> Optional[StorePlugin]:
|
||||
"""
|
||||
获取插件信息
|
||||
|
||||
Args:
|
||||
plugin_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_name:
|
||||
return plugin
|
||||
return None
|
||||
|
||||
|
||||
def get_plugin_default_enable(plugin_name: str) -> bool:
|
||||
"""
|
||||
获取插件默认启用状态，由插件定义，不存在则默认为启用，优先从缓存中获取
|
||||
|
||||
Args:
|
||||
plugin_name (str): 插件模块名
|
||||
|
||||
Returns:
|
||||
bool: 插件默认状态
|
||||
"""
|
||||
if plugin_name not in __default_enable:
|
||||
plug = nonebot.plugin.get_plugin(plugin_name)
|
||||
default_enable = (plug.metadata.extra.get("default_enable", True) if plug.metadata else True) if plug else True
|
||||
__default_enable[plugin_name] = default_enable
|
||||
|
||||
return __default_enable[plugin_name]
|
||||
|
||||
|
||||
def get_plugin_session_enable(event: T_MessageEvent, plugin_name: str) -> bool:
|
||||
"""
|
||||
获取插件当前会话启用状态
|
||||
|
||||
Args:
|
||||
event: 会话事件
|
||||
plugin_name (str): 插件模块名
|
||||
|
||||
Returns:
|
||||
bool: 插件当前状态
|
||||
"""
|
||||
if isinstance(event, satori.event.Event):
|
||||
if event.guild is not None:
|
||||
message_type = "group"
|
||||
else:
|
||||
message_type = "private"
|
||||
else:
|
||||
message_type = event.message_type
|
||||
if message_type == "group":
|
||||
group_id = str(event.guild.id if isinstance(event, satori.event.Event) else event.group_id)
|
||||
if group_id not in __group_data:
|
||||
group: Group = group_db.where_one(Group(), "group_id = ?", group_id, default=Group(group_id=group_id))
|
||||
__group_data[str(group_id)] = group
|
||||
|
||||
session = __group_data[group_id]
|
||||
else:
|
||||
# session: User = user_db.first(User(), "user_id = ?", event.user_id, default=User(user_id=str(event.user_id)))
|
||||
user_id = str(event.user.id if isinstance(event, satori.event.Event) else event.user_id)
|
||||
if user_id not in __user_data:
|
||||
user: User = user_db.where_one(User(), "user_id = ?", user_id, default=User(user_id=user_id))
|
||||
__user_data[user_id] = user
|
||||
session = __user_data[user_id]
|
||||
# 默认停用插件在启用列表内表示启用
|
||||
# 默认停用插件不在启用列表内表示停用
|
||||
# 默认启用插件在停用列表内表示停用
|
||||
# 默认启用插件不在停用列表内表示启用
|
||||
default_enable = get_plugin_default_enable(plugin_name)
|
||||
if default_enable:
|
||||
return plugin_name not in session.disabled_plugins
|
||||
else:
|
||||
return plugin_name in session.enabled_plugins
|
||||
|
||||
|
||||
def set_plugin_session_enable(event: T_MessageEvent, plugin_name: str, enable: bool):
|
||||
"""
|
||||
设置插件会话启用状态，同时更新数据库和缓存
|
||||
Args:
|
||||
event:
|
||||
plugin_name:
|
||||
enable:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if event_utils.get_message_type(event) == "group":
|
||||
session: Group = group_db.where_one(Group(), "group_id = ?", str(event_utils.get_group_id(event)),
|
||||
default=Group(group_id=str(event_utils.get_group_id(event))))
|
||||
else:
|
||||
session: User = user_db.where_one(User(), "user_id = ?", str(event_utils.get_user_id(event)),
|
||||
default=User(user_id=str(event_utils.get_user_id(event))))
|
||||
default_enable = get_plugin_default_enable(plugin_name)
|
||||
if default_enable:
|
||||
if enable:
|
||||
session.disabled_plugins.remove(plugin_name)
|
||||
else:
|
||||
session.disabled_plugins.append(plugin_name)
|
||||
else:
|
||||
if enable:
|
||||
session.enabled_plugins.append(plugin_name)
|
||||
else:
|
||||
session.enabled_plugins.remove(plugin_name)
|
||||
|
||||
if event_utils.get_message_type(event) == "group":
|
||||
__group_data[str(event_utils.get_group_id(event))] = session
|
||||
group_db.save(session)
|
||||
else:
|
||||
__user_data[str(event_utils.get_user_id(event))] = session
|
||||
user_db.save(session)
|
||||
|
||||
|
||||
def get_plugin_global_enable(plugin_name: str) -> bool:
|
||||
"""
|
||||
获取插件全局启用状态, 优先从缓存中获取
|
||||
Args:
|
||||
plugin_name:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if plugin_name not in __global_enable:
|
||||
plugin = plugin_db.where_one(
|
||||
GlobalPlugin(),
|
||||
"module_name = ?",
|
||||
plugin_name,
|
||||
default=GlobalPlugin(module_name=plugin_name, enabled=True))
|
||||
__global_enable[plugin_name] = plugin.enabled
|
||||
|
||||
return __global_enable[plugin_name]
|
||||
|
||||
|
||||
def set_plugin_global_enable(plugin_name: str, enable: bool):
|
||||
"""
|
||||
设置插件全局启用状态，同时更新数据库和缓存
|
||||
Args:
|
||||
plugin_name:
|
||||
enable:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
plugin = plugin_db.where_one(
|
||||
GlobalPlugin(),
|
||||
"module_name = ?",
|
||||
plugin_name,
|
||||
default=GlobalPlugin(module_name=plugin_name, enabled=True))
|
||||
plugin.enabled = enable
|
||||
|
||||
plugin_db.save(plugin)
|
||||
__global_enable[plugin_name] = enable
|
||||
|
||||
|
||||
def get_plugin_can_be_toggle(plugin_name: str) -> bool:
|
||||
"""
|
||||
获取插件是否可以被启用/停用
|
||||
|
||||
Args:
|
||||
plugin_name (str): 插件模块名
|
||||
|
||||
Returns:
|
||||
bool: 插件是否可以被启用/停用
|
||||
"""
|
||||
plug = nonebot.plugin.get_plugin(plugin_name)
|
||||
return plug.metadata.extra.get("toggleable", True) if plug and plug.metadata else True
|
||||
|
||||
|
||||
def get_group_enable(group_id: str) -> bool:
|
||||
"""
|
||||
获取群组是否启用插机器人
|
||||
|
||||
Args:
|
||||
group_id (str): 群组ID
|
||||
|
||||
Returns:
|
||||
bool: 群组是否启用插件
|
||||
"""
|
||||
group_id = str(group_id)
|
||||
if group_id not in __group_data:
|
||||
group: Group = group_db.where_one(Group(), "group_id = ?", group_id, default=Group(group_id=group_id))
|
||||
__group_data[group_id] = group
|
||||
|
||||
return __group_data[group_id].enable
|
||||
|
||||
|
||||
def set_group_enable(group_id: str, enable: bool):
|
||||
"""
|
||||
设置群组是否启用插机器人
|
||||
|
||||
Args:
|
||||
group_id (str): 群组ID
|
||||
enable (bool): 是否启用
|
||||
"""
|
||||
group_id = str(group_id)
|
||||
group: Group = group_db.where_one(Group(), "group_id = ?", group_id, default=Group(group_id=group_id))
|
||||
group.enable = enable
|
||||
|
||||
__group_data[group_id] = group
|
||||
group_db.save(group)
|
||||
846
src/nonebot_plugins/liteyuki_pacman/npm.py
Normal file
846
src/nonebot_plugins/liteyuki_pacman/npm.py
Normal file
@ -0,0 +1,846 @@
|
||||
import os
|
||||
import sys
|
||||
import aiohttp
|
||||
import nonebot.plugin
|
||||
import pip
|
||||
from io import StringIO
|
||||
from arclet.alconna import MultiVar
|
||||
from nonebot import Bot, require
|
||||
from nonebot.exception import FinishedException, IgnoredException, MockApiException
|
||||
from nonebot.internal.adapter import Event
|
||||
from nonebot.internal.matcher import Matcher
|
||||
from nonebot.message import run_preprocessor
|
||||
from nonebot.permission import SUPERUSER
|
||||
from nonebot.plugin import Plugin, PluginMetadata
|
||||
from nonebot.utils import run_sync
|
||||
|
||||
from src.utils.base.data_manager import InstalledPlugin
|
||||
from src.utils.base.language import get_user_lang
|
||||
from src.utils.base.ly_typing import T_Bot
|
||||
from src.utils.message.message import MarkdownMessage as md
|
||||
from src.utils.message.markdown import MarkdownComponent as mdc, compile_md, escape_md
|
||||
from src.utils.base.permission import GROUP_ADMIN, GROUP_OWNER
|
||||
from src.utils.message.tools import clamp
|
||||
from .common import *
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
from nonebot_plugin_alconna import (
|
||||
on_alconna,
|
||||
Alconna,
|
||||
Args,
|
||||
Arparma,
|
||||
Subcommand,
|
||||
Option,
|
||||
OptionResult,
|
||||
SubcommandResult,
|
||||
)
|
||||
|
||||
# const
|
||||
enable_global = "enable-global"
|
||||
disable_global = "disable-global"
|
||||
enable = "enable"
|
||||
disable = "disable"
|
||||
|
||||
|
||||
@on_alconna(
|
||||
aliases={"插件"},
|
||||
command=Alconna(
|
||||
"npm",
|
||||
Subcommand(
|
||||
"enable",
|
||||
Args["plugin_name", str],
|
||||
Option(
|
||||
"-g|--group",
|
||||
Args["group_id", str, None],
|
||||
help_text="群号",
|
||||
),
|
||||
alias=["e", "启用"],
|
||||
),
|
||||
Subcommand(
|
||||
"disable",
|
||||
Args["plugin_name", str],
|
||||
Option(
|
||||
"-g|--group",
|
||||
Args["group_id", str, None],
|
||||
help_text="群号",
|
||||
),
|
||||
alias=["d", "停用"],
|
||||
),
|
||||
Subcommand(
|
||||
enable_global,
|
||||
Args["plugin_name", str],
|
||||
alias=["eg", "全局启用"],
|
||||
),
|
||||
Subcommand(
|
||||
disable_global,
|
||||
Args["plugin_name", str],
|
||||
alias=["dg", "全局停用"],
|
||||
),
|
||||
# 安装部分
|
||||
Subcommand(
|
||||
"update",
|
||||
alias=["u", "更新"],
|
||||
),
|
||||
Subcommand(
|
||||
"search",
|
||||
Args["keywords", MultiVar(str)],
|
||||
alias=["s", "搜索"],
|
||||
),
|
||||
Subcommand(
|
||||
"install",
|
||||
Args["plugin_name", str],
|
||||
alias=["i", "安装"],
|
||||
),
|
||||
Subcommand(
|
||||
"uninstall",
|
||||
Args["plugin_name", str],
|
||||
alias=["r", "rm", "卸载"],
|
||||
),
|
||||
Subcommand(
|
||||
"list",
|
||||
Args["page", int, 1]["num", int, 10],
|
||||
alias=["ls", "列表"],
|
||||
),
|
||||
),
|
||||
).handle()
|
||||
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, npm: Matcher):
|
||||
if not os.path.exists("data/liteyuki/plugins.json"):
|
||||
await npm_update()
|
||||
# 判断会话类型
|
||||
ulang = get_user_lang(str(event.user_id))
|
||||
plugin_name = result.args.get("plugin_name")
|
||||
sc = result.subcommands # 获取子命令
|
||||
perm_s = await SUPERUSER(bot, event) # 判断是否为超级用户
|
||||
# 支持对自定义command_start的判断
|
||||
if sc.get("enable") or sc.get("disable"):
|
||||
|
||||
toggle = result.subcommands.get("enable") is not None
|
||||
|
||||
plugin_exist = get_plugin_exist(plugin_name)
|
||||
|
||||
# 判定会话类型
|
||||
# 输入群号
|
||||
if (
|
||||
group_id := (
|
||||
sc.get("enable", SubcommandResult())
|
||||
.options.get("group", OptionResult())
|
||||
.args.get("group_id")
|
||||
or sc.get("disable", SubcommandResult())
|
||||
.options.get("group", OptionResult())
|
||||
.args.get("group_id")
|
||||
)
|
||||
) and await SUPERUSER(bot, event):
|
||||
session_id = group_id
|
||||
new_event = event.copy()
|
||||
new_event.group_id = group_id
|
||||
new_event.message_type = "group"
|
||||
|
||||
elif event.message_type == "private":
|
||||
session_id = event.user_id
|
||||
new_event = event
|
||||
else:
|
||||
if (
|
||||
await GROUP_ADMIN(bot, event)
|
||||
or await GROUP_OWNER(bot, event)
|
||||
or await SUPERUSER(bot, event)
|
||||
):
|
||||
session_id = event.group_id
|
||||
new_event = event
|
||||
else:
|
||||
raise FinishedException(ulang.get("Permission Denied"))
|
||||
|
||||
session_enable = get_plugin_session_enable(
|
||||
new_event, plugin_name
|
||||
) # 获取插件当前状态
|
||||
|
||||
can_be_toggled = get_plugin_can_be_toggle(
|
||||
plugin_name
|
||||
) # 获取插件是否可以被启用/停用
|
||||
|
||||
if not plugin_exist:
|
||||
await npm.finish(ulang.get("npm.plugin_not_found", NAME=plugin_name))
|
||||
|
||||
if not can_be_toggled:
|
||||
await npm.finish(
|
||||
ulang.get("npm.plugin_cannot_be_toggled", NAME=plugin_name)
|
||||
)
|
||||
|
||||
if session_enable == toggle:
|
||||
await npm.finish(
|
||||
ulang.get(
|
||||
"npm.plugin_already",
|
||||
NAME=plugin_name,
|
||||
STATUS=(
|
||||
ulang.get("npm.enable") if toggle else ulang.get("npm.disable")
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# 键入自定义群号的情况
|
||||
|
||||
try:
|
||||
set_plugin_session_enable(new_event, plugin_name, toggle)
|
||||
except Exception as e:
|
||||
nonebot.logger.error(e)
|
||||
await npm.finish(
|
||||
ulang.get(
|
||||
"npm.toggle_failed",
|
||||
NAME=plugin_name,
|
||||
STATUS=(
|
||||
ulang.get("npm.enable") if toggle else ulang.get("npm.disable")
|
||||
),
|
||||
ERROR=str(e),
|
||||
)
|
||||
)
|
||||
|
||||
await npm.finish(
|
||||
ulang.get(
|
||||
"npm.toggle_success",
|
||||
NAME=plugin_name,
|
||||
STATUS=(
|
||||
ulang.get("npm.enable") if toggle else ulang.get("npm.disable")
|
||||
),
|
||||
) # + str(session_id) 这里应该不需增加一个id，在任何语言文件里，这句话都不是这样翻的，你是不是调试的时候忘删了？
|
||||
)
|
||||
|
||||
elif (
|
||||
sc.get(enable_global)
|
||||
or result.subcommands.get(disable_global)
|
||||
and await SUPERUSER(bot, event)
|
||||
):
|
||||
plugin_exist = get_plugin_exist(plugin_name)
|
||||
|
||||
toggle = result.subcommands.get(enable_global) is not None
|
||||
|
||||
can_be_toggled = get_plugin_can_be_toggle(plugin_name)
|
||||
|
||||
if not plugin_exist:
|
||||
await npm.finish(ulang.get("npm.plugin_not_found", NAME=plugin_name))
|
||||
|
||||
if not can_be_toggled:
|
||||
await npm.finish(
|
||||
ulang.get("npm.plugin_cannot_be_toggled", NAME=plugin_name)
|
||||
)
|
||||
|
||||
global_enable = get_plugin_global_enable(plugin_name)
|
||||
if global_enable == toggle:
|
||||
await npm.finish(
|
||||
ulang.get(
|
||||
"npm.plugin_already",
|
||||
NAME=plugin_name,
|
||||
STATUS=(
|
||||
ulang.get("npm.enable") if toggle else ulang.get("npm.disable")
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
try:
|
||||
set_plugin_global_enable(plugin_name, toggle)
|
||||
except Exception as e:
|
||||
await npm.finish(
|
||||
ulang.get(
|
||||
"npm.toggle_failed",
|
||||
NAME=plugin_name,
|
||||
STATUS=(
|
||||
ulang.get("npm.enable") if toggle else ulang.get("npm.disable")
|
||||
),
|
||||
ERROR=str(e),
|
||||
)
|
||||
)
|
||||
|
||||
await npm.finish(
|
||||
ulang.get(
|
||||
"npm.toggle_success",
|
||||
NAME=plugin_name,
|
||||
STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"),
|
||||
)
|
||||
)
|
||||
|
||||
elif sc.get("update") and perm_s:
|
||||
r = await npm_update()
|
||||
if r:
|
||||
await npm.finish(ulang.get("npm.store_update_success"))
|
||||
else:
|
||||
await npm.finish(ulang.get("npm.store_update_failed"))
|
||||
|
||||
elif sc.get("search"):
|
||||
keywords: list[str] = result.subcommands["search"].args.get("keywords")
|
||||
rs = await npm_search(keywords)
|
||||
max_show = 10
|
||||
if len(rs):
|
||||
reply = f"{ulang.get('npm.search_result')} | {ulang.get('npm.total', TOTAL=len(rs))}\n***"
|
||||
for storePlugin in rs[: min(max_show, len(rs))]:
|
||||
btn_install_or_update = md.btn_cmd(
|
||||
(
|
||||
ulang.get("npm.update")
|
||||
if get_plugin_exist(storePlugin.module_name)
|
||||
else ulang.get("npm.install")
|
||||
),
|
||||
"npm install %s" % storePlugin.module_name,
|
||||
)
|
||||
link_page = md.btn_link(ulang.get("npm.homepage"), storePlugin.homepage)
|
||||
link_pypi = md.btn_link(ulang.get("npm.pypi"), storePlugin.homepage)
|
||||
|
||||
reply += (
|
||||
f"\n# **{storePlugin.name}**\n"
|
||||
f"\n> **{storePlugin.desc}**\n"
|
||||
f"\n> {ulang.get('npm.author')}: {storePlugin.author}"
|
||||
f"\n> *{md.escape(storePlugin.module_name)}*"
|
||||
f"\n> {btn_install_or_update} {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:
|
||||
reply = ulang.get("npm.search_no_result")
|
||||
await md.send_md(reply, bot, event=event)
|
||||
|
||||
elif sc.get("install") and perm_s:
|
||||
plugin_name: str = result.subcommands["install"].args.get("plugin_name")
|
||||
store_plugin = await get_store_plugin(plugin_name)
|
||||
await npm.send(ulang.get("npm.installing", NAME=plugin_name))
|
||||
|
||||
r, log = await npm_install(plugin_name)
|
||||
log = log.replace("\\", "/")
|
||||
|
||||
if not store_plugin:
|
||||
await npm.finish(ulang.get("npm.plugin_not_found", NAME=plugin_name))
|
||||
|
||||
homepage_btn = md.btn_cmd(ulang.get("npm.homepage"), store_plugin.homepage)
|
||||
if r:
|
||||
r_load = nonebot.load_plugin(plugin_name) # 加载插件
|
||||
installed_plugin = InstalledPlugin(
|
||||
module_name=plugin_name
|
||||
) # 构造插件信息模型
|
||||
found_in_db_plugin = plugin_db.where_one(
|
||||
InstalledPlugin(), "module_name = ?", plugin_name
|
||||
) # 查询数据库中是否已经安装
|
||||
if r_load:
|
||||
if found_in_db_plugin is None:
|
||||
plugin_db.save(installed_plugin)
|
||||
info = md.escape(
|
||||
ulang.get("npm.install_success", NAME=store_plugin.name)
|
||||
) # markdown转义
|
||||
await md.send_md(f"{info}\n\n" f"```\n{log}\n```", bot, event=event)
|
||||
else:
|
||||
await npm.finish(
|
||||
ulang.get(
|
||||
"npm.plugin_already_installed", NAME=store_plugin.name
|
||||
)
|
||||
)
|
||||
else:
|
||||
info = ulang.get(
|
||||
"npm.load_failed", NAME=plugin_name, HOMEPAGE=homepage_btn
|
||||
).replace("_", r"\\_")
|
||||
await md.send_md(f"{info}\n\n" f"```\n{log}\n```\n", bot, event=event)
|
||||
else:
|
||||
info = ulang.get(
|
||||
"npm.install_failed", NAME=plugin_name, HOMEPAGE=homepage_btn
|
||||
).replace("_", r"\\_")
|
||||
await md.send_md(f"{info}\n\n" f"```\n{log}\n```", bot, event=event)
|
||||
|
||||
elif sc.get("uninstall") and perm_s:
|
||||
plugin_name: str = result.subcommands["uninstall"].args.get("plugin_name") # type: ignore
|
||||
found_installed_plugin: InstalledPlugin = plugin_db.where_one(
|
||||
InstalledPlugin(), "module_name = ?", plugin_name
|
||||
)
|
||||
if found_installed_plugin:
|
||||
plugin_db.delete(InstalledPlugin(), "module_name = ?", plugin_name)
|
||||
reply = f"{ulang.get('npm.uninstall_success', NAME=found_installed_plugin.module_name)}"
|
||||
await npm.finish(reply)
|
||||
else:
|
||||
await npm.finish(ulang.get("npm.plugin_not_installed", NAME=plugin_name))
|
||||
|
||||
elif sc.get("list"):
|
||||
loaded_plugin_list = sorted(nonebot.get_loaded_plugins(), key=lambda x: x.name)
|
||||
num_per_page = result.subcommands.get("list").args.get("num")
|
||||
total = len(loaded_plugin_list) // num_per_page + (
|
||||
1 if len(loaded_plugin_list) % num_per_page else 0
|
||||
)
|
||||
|
||||
page = clamp(result.subcommands.get("list").args.get("page"), 1, total)
|
||||
|
||||
# 已加载插件 | 总计10 | 第1/3页
|
||||
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"
|
||||
)
|
||||
|
||||
permission_oas = (
|
||||
await GROUP_ADMIN(bot, event)
|
||||
or await GROUP_OWNER(bot, event)
|
||||
or await SUPERUSER(bot, event)
|
||||
)
|
||||
permission_s = await SUPERUSER(bot, event)
|
||||
|
||||
for storePlugin in loaded_plugin_list[
|
||||
(page - 1)
|
||||
* num_per_page : min(page * num_per_page, len(loaded_plugin_list))
|
||||
]:
|
||||
# 检查是否有 metadata 属性
|
||||
# 添加帮助按钮
|
||||
|
||||
btn_usage = md.btn_cmd(
|
||||
ulang.get("npm.usage"), f"help {storePlugin.name}", False
|
||||
)
|
||||
store_plugin = await get_store_plugin(storePlugin.name)
|
||||
session_enable = get_plugin_session_enable(event, storePlugin.name)
|
||||
if store_plugin:
|
||||
# btn_homepage = md.btn_link(ulang.get("npm.homepage"), store_plugin.homepage)
|
||||
show_name = store_plugin.name
|
||||
elif storePlugin.metadata:
|
||||
# if storePlugin.metadata.extra.get("liteyuki"):
|
||||
# btn_homepage = md.btn_link(ulang.get("npm.homepage"), "https://github.com/snowykami/LiteyukiBot")
|
||||
# else:
|
||||
# btn_homepage = ulang.get("npm.homepage")
|
||||
show_name = storePlugin.metadata.name
|
||||
else:
|
||||
# btn_homepage = ulang.get("npm.homepage")
|
||||
show_name = storePlugin.name
|
||||
ulang.get("npm.no_description")
|
||||
|
||||
if storePlugin.metadata:
|
||||
reply += f"\n**{md.escape(show_name)}**\n"
|
||||
else:
|
||||
reply += f"**{md.escape(show_name)}**\n"
|
||||
|
||||
reply += f"\n > {btn_usage}"
|
||||
|
||||
if permission_oas:
|
||||
# 添加启用/停用插件按钮
|
||||
cmd_toggle = f"npm {'disable' if session_enable else 'enable'} {storePlugin.name}"
|
||||
text_toggle = ulang.get(
|
||||
"npm.disable" if session_enable else "npm.enable"
|
||||
)
|
||||
can_be_toggle = get_plugin_can_be_toggle(storePlugin.name)
|
||||
btn_toggle = (
|
||||
text_toggle
|
||||
if not can_be_toggle
|
||||
else md.btn_cmd(text_toggle, cmd_toggle)
|
||||
)
|
||||
reply += f" {btn_toggle}"
|
||||
|
||||
if permission_s:
|
||||
plugin_in_database = plugin_db.where_one(
|
||||
InstalledPlugin(), "module_name = ?", storePlugin.name
|
||||
)
|
||||
# 添加移除插件和全局切换按钮
|
||||
global_enable = get_plugin_global_enable(storePlugin.name)
|
||||
btn_uninstall = (
|
||||
(
|
||||
md.btn_cmd(
|
||||
ulang.get("npm.uninstall"),
|
||||
f"npm uninstall {storePlugin.name}",
|
||||
)
|
||||
)
|
||||
if plugin_in_database
|
||||
else ulang.get("npm.uninstall")
|
||||
)
|
||||
btn_toggle_global_text = ulang.get(
|
||||
"npm.disable_global" if global_enable else "npm.enable_global"
|
||||
)
|
||||
cmd_toggle_global = f"npm {'disable' if global_enable else 'enable'}-global {storePlugin.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"
|
||||
# 根据页数添加翻页按钮。第一页显示上一页文本而不是按钮，最后一页显示下一页文本而不是按钮
|
||||
btn_prev = (
|
||||
md.btn_cmd(
|
||||
ulang.get("npm.prev_page"), f"npm list {page - 1} {num_per_page}"
|
||||
)
|
||||
if page > 1
|
||||
else ulang.get("npm.prev_page")
|
||||
)
|
||||
btn_next = (
|
||||
md.btn_cmd(
|
||||
ulang.get("npm.next_page"), f"npm list {page + 1} {num_per_page}"
|
||||
)
|
||||
if page < total
|
||||
else ulang.get("npm.next_page")
|
||||
)
|
||||
reply += f"\n{btn_prev} {page}/{total} {btn_next}"
|
||||
await md.send_md(reply, bot, event=event)
|
||||
|
||||
else:
|
||||
if await SUPERUSER(bot, event):
|
||||
btn_enable_global = md.btn_cmd(
|
||||
ulang.get("npm.enable_global"), "npm enable-global", False, False
|
||||
)
|
||||
btn_disable_global = md.btn_cmd(
|
||||
ulang.get("npm.disable_global"), "npm disable-global", False, False
|
||||
)
|
||||
btn_search = md.btn_cmd(
|
||||
ulang.get("npm.search"), "npm search ", False, False
|
||||
)
|
||||
btn_uninstall_ = md.btn_cmd(
|
||||
ulang.get("npm.uninstall"), "npm uninstall ", False, False
|
||||
)
|
||||
btn_install_ = md.btn_cmd(
|
||||
ulang.get("npm.install"), "npm install ", False, False
|
||||
)
|
||||
btn_update = md.btn_cmd(
|
||||
ulang.get("npm.update_index"), "npm update", False, True
|
||||
)
|
||||
btn_list = md.btn_cmd(
|
||||
ulang.get("npm.list_plugins"), "npm list ", False, False
|
||||
)
|
||||
btn_disable = md.btn_cmd(
|
||||
ulang.get("npm.disable_session"), "npm disable ", False, False
|
||||
)
|
||||
btn_enable = md.btn_cmd(
|
||||
ulang.get("npm.enable_session"), "npm enable ", False, False
|
||||
)
|
||||
reply = (
|
||||
f"\n# **{ulang.get('npm.help')}**"
|
||||
f"\n{btn_update}"
|
||||
f"\n\n>*{md.escape('npm update')}*\n"
|
||||
f"\n{btn_install_}"
|
||||
f"\n\n>*{md.escape('npm install <plugin_name')}*>\n"
|
||||
f"\n{btn_uninstall_}"
|
||||
f"\n\n>*{md.escape('npm uninstall <plugin_name')}*>\n"
|
||||
f"\n{btn_search}"
|
||||
f"\n\n>*{md.escape('npm search <keywords...')}*>\n"
|
||||
f"\n{btn_disable_global}"
|
||||
f"\n\n>*{md.escape('npm disable-global <plugin_name')}*>\n"
|
||||
f"\n{btn_enable_global}"
|
||||
f"\n\n>*{md.escape('npm enable-global <plugin_name')}*>\n"
|
||||
f"\n{btn_disable}"
|
||||
f"\n\n>*{md.escape('npm disable <plugin_name')}*>\n"
|
||||
f"\n{btn_enable}"
|
||||
f"\n\n>*{md.escape('npm enable <plugin_name')}*>\n"
|
||||
f"\n{btn_list}"
|
||||
f"\n\n>page为页数，num为每页显示数量"
|
||||
f"\n\n>*{md.escape('npm list [page] [num]')}*"
|
||||
)
|
||||
await md.send_md(reply, bot, event=event)
|
||||
else:
|
||||
|
||||
btn_list = md.btn_cmd(
|
||||
ulang.get("npm.list_plugins"), "npm list ", False, False
|
||||
)
|
||||
btn_disable = md.btn_cmd(
|
||||
ulang.get("npm.disable_session"), "npm disable ", False, False
|
||||
)
|
||||
btn_enable = md.btn_cmd(
|
||||
ulang.get("npm.enable_session"), "npm enable ", False, False
|
||||
)
|
||||
reply = (
|
||||
f"\n# **{ulang.get('npm.help')}**"
|
||||
f"\n{btn_disable}"
|
||||
f"\n\n>*{md.escape('npm disable <plugin_name')}*>\n"
|
||||
f"\n{btn_enable}"
|
||||
f"\n\n>*{md.escape('npm enable <plugin_name')}*>\n"
|
||||
f"\n{btn_list}"
|
||||
f"\n\n>page为页数，num为每页显示数量"
|
||||
f"\n\n>*{md.escape('npm list [page] [num]')}*"
|
||||
)
|
||||
await md.send_md(reply, bot, event=event)
|
||||
|
||||
|
||||
@on_alconna(
|
||||
aliases={"群聊"},
|
||||
command=Alconna(
|
||||
"gm",
|
||||
Subcommand(
|
||||
enable,
|
||||
Args["group_id", str, None],
|
||||
alias=["e", "启用"],
|
||||
),
|
||||
Subcommand(
|
||||
disable,
|
||||
Args["group_id", str, None],
|
||||
alias=["d", "停用"],
|
||||
),
|
||||
),
|
||||
permission=SUPERUSER | GROUP_OWNER | GROUP_ADMIN,
|
||||
).handle()
|
||||
async def _(bot: T_Bot, event: T_MessageEvent, gm: Matcher, result: Arparma):
|
||||
ulang = get_user_lang(str(event.user_id))
|
||||
to_enable = result.subcommands.get(enable) is not None
|
||||
|
||||
group_id = None
|
||||
if await SUPERUSER(bot, event):
|
||||
# 仅超级用户可以自定义群号
|
||||
group_id = result.subcommands.get(
|
||||
enable, result.subcommands.get(disable)
|
||||
).args.get("group_id")
|
||||
if group_id is None and event.message_type == "group":
|
||||
group_id = str(event.group_id)
|
||||
|
||||
if group_id is None:
|
||||
await gm.finish(ulang.get("liteyuki.invalid_command"), liteyuki_pass=True)
|
||||
|
||||
enabled = get_group_enable(group_id)
|
||||
if enabled == to_enable:
|
||||
await gm.finish(
|
||||
ulang.get(
|
||||
"liteyuki.group_already",
|
||||
STATUS=(
|
||||
ulang.get("npm.enable") if to_enable else ulang.get("npm.disable")
|
||||
),
|
||||
GROUP=group_id,
|
||||
),
|
||||
liteyuki_pass=True,
|
||||
)
|
||||
else:
|
||||
set_group_enable(group_id, to_enable)
|
||||
await gm.finish(
|
||||
ulang.get(
|
||||
"liteyuki.group_success",
|
||||
STATUS=(
|
||||
ulang.get("npm.enable") if to_enable else ulang.get("npm.disable")
|
||||
),
|
||||
GROUP=group_id,
|
||||
),
|
||||
liteyuki_pass=True,
|
||||
)
|
||||
|
||||
|
||||
@on_alconna(
|
||||
aliases={"帮助"},
|
||||
command=Alconna(
|
||||
"help",
|
||||
Args["plugin_name", str, None],
|
||||
),
|
||||
).handle()
|
||||
async def _(result: Arparma, matcher: Matcher, event: T_MessageEvent, bot: T_Bot):
|
||||
ulang = get_user_lang(str(event.user_id))
|
||||
plugin_name = result.main_args.get("plugin_name")
|
||||
if plugin_name:
|
||||
searched_plugins = search_loaded_plugin(plugin_name)
|
||||
if searched_plugins:
|
||||
loaded_plugin = searched_plugins[0]
|
||||
else:
|
||||
await matcher.finish(ulang.get("npm.plugin_not_found", NAME=plugin_name))
|
||||
|
||||
if loaded_plugin:
|
||||
if loaded_plugin.metadata is None:
|
||||
loaded_plugin.metadata = PluginMetadata(
|
||||
name=plugin_name, description="", usage=""
|
||||
)
|
||||
# 从商店获取详细信息
|
||||
store_plugin = await get_store_plugin(plugin_name)
|
||||
if loaded_plugin.metadata.extra.get("liteyuki"):
|
||||
store_plugin = StorePlugin(
|
||||
name=loaded_plugin.metadata.name,
|
||||
desc=loaded_plugin.metadata.description,
|
||||
author="SnowyKami",
|
||||
module_name=plugin_name,
|
||||
homepage="https://github.com/snowykami/LiteyukiBot",
|
||||
)
|
||||
elif store_plugin is None:
|
||||
store_plugin = StorePlugin(
|
||||
name=loaded_plugin.metadata.name,
|
||||
desc=loaded_plugin.metadata.description,
|
||||
author="",
|
||||
module_name=plugin_name,
|
||||
homepage="",
|
||||
)
|
||||
|
||||
if store_plugin:
|
||||
link = store_plugin.homepage
|
||||
elif loaded_plugin.metadata.extra.get("liteyuki"):
|
||||
link = "https://github.com/snowykami/LiteyukiBot"
|
||||
else:
|
||||
link = None
|
||||
|
||||
reply = [
|
||||
mdc.heading(escape_md(store_plugin.name)),
|
||||
mdc.quote(store_plugin.module_name),
|
||||
mdc.quote(
|
||||
mdc.bold(ulang.get("npm.author"))
|
||||
+ " "
|
||||
+ (
|
||||
mdc.link(
|
||||
store_plugin.author,
|
||||
f"https://github.com/{store_plugin.author}",
|
||||
)
|
||||
if store_plugin.author
|
||||
else "Unknown"
|
||||
)
|
||||
),
|
||||
mdc.quote(
|
||||
mdc.bold(ulang.get("npm.description"))
|
||||
+ " "
|
||||
+ mdc.paragraph(
|
||||
max(loaded_plugin.metadata.description, store_plugin.desc)
|
||||
)
|
||||
),
|
||||
mdc.heading(ulang.get("npm.usage"), 2),
|
||||
mdc.paragraph(loaded_plugin.metadata.usage.replace("\n", "\n\n")),
|
||||
(
|
||||
mdc.link(ulang.get("npm.homepage"), link)
|
||||
if link
|
||||
else mdc.paragraph(ulang.get("npm.homepage"))
|
||||
),
|
||||
]
|
||||
await md.send_md(compile_md(reply), bot, event=event)
|
||||
else:
|
||||
await matcher.finish(ulang.get("npm.plugin_not_found", NAME=plugin_name))
|
||||
else:
|
||||
pass
|
||||
|
||||
|
||||
# 传入事件阻断hook
|
||||
@run_preprocessor
|
||||
async def pre_handle(event: Event, matcher: Matcher):
|
||||
plugin: Plugin = matcher.plugin
|
||||
plugin_global_enable = get_plugin_global_enable(plugin.name)
|
||||
if not plugin_global_enable:
|
||||
raise IgnoredException("Plugin disabled globally")
|
||||
if event.get_type() == "message":
|
||||
plugin_session_enable = get_plugin_session_enable(event, plugin.name)
|
||||
if not plugin_session_enable:
|
||||
raise IgnoredException("Plugin disabled in session")
|
||||
|
||||
|
||||
# 群聊开关阻断hook
|
||||
@Bot.on_calling_api
|
||||
async def block_disable_session(bot: Bot, api: str, args: dict):
|
||||
if "group_id" in args and not args.get("liteyuki_pass", False):
|
||||
group_id = args["group_id"]
|
||||
if not get_group_enable(group_id):
|
||||
nonebot.logger.debug(f"Group {group_id} disabled")
|
||||
raise MockApiException(f"Group {group_id} disabled")
|
||||
|
||||
|
||||
async def npm_update() -> bool:
|
||||
"""
|
||||
更新本地插件json缓存
|
||||
|
||||
Returns:
|
||||
bool: 是否成功更新
|
||||
"""
|
||||
url_list = [
|
||||
"https://registry.nonebot.dev/plugins.json",
|
||||
]
|
||||
for url in url_list:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
if resp.status == 200:
|
||||
async with aiofiles.open("data/liteyuki/plugins.json", "wb") as f:
|
||||
data = await resp.read()
|
||||
await f.write(data)
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
async def npm_search(keywords: list[str]) -> list[StorePlugin]:
|
||||
"""
|
||||
在本地缓存商店数据中搜索插件
|
||||
|
||||
Args:
|
||||
keywords (list[str]): 关键词列表
|
||||
|
||||
Returns:
|
||||
list[StorePlugin]: 插件列表
|
||||
"""
|
||||
plugin_blacklist = [
|
||||
"nonebot_plugin_xiuxian_2",
|
||||
"nonebot_plugin_htmlrender",
|
||||
"nonebot_plugin_alconna",
|
||||
]
|
||||
|
||||
results = []
|
||||
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 in plugin_blacklist:
|
||||
continue
|
||||
plugin_text = " ".join(
|
||||
[
|
||||
plugin.name,
|
||||
plugin.desc,
|
||||
plugin.author,
|
||||
plugin.module_name,
|
||||
" ".join([tag.label for tag in plugin.tags]),
|
||||
]
|
||||
)
|
||||
if all([keyword in plugin_text for keyword in keywords]):
|
||||
results.append(plugin)
|
||||
return results
|
||||
|
||||
|
||||
@run_sync
|
||||
def npm_install(plugin_package_name) -> tuple[bool, str]:
|
||||
"""
|
||||
异步安装插件，使用pip安装
|
||||
Args:
|
||||
plugin_package_name:
|
||||
|
||||
Returns:
|
||||
tuple[bool, str]: 是否成功，输出信息
|
||||
|
||||
"""
|
||||
# 重定向标准输出
|
||||
buffer = StringIO()
|
||||
sys.stdout = buffer
|
||||
sys.stderr = buffer
|
||||
|
||||
update = False
|
||||
if get_plugin_exist(plugin_package_name):
|
||||
update = True
|
||||
|
||||
mirrors = [
|
||||
"https://pypi.tuna.tsinghua.edu.cn/simple", # 清华大学
|
||||
"https://pypi.org/simple", # 官方源
|
||||
]
|
||||
|
||||
# 使用pip安装包，对每个镜像尝试一次，成功后返回值
|
||||
success = False
|
||||
for mirror in mirrors:
|
||||
try:
|
||||
nonebot.logger.info(f"pip install try mirror: {mirror}")
|
||||
if update:
|
||||
result = pip.main(
|
||||
["install", "--upgrade", plugin_package_name, "-i", mirror]
|
||||
)
|
||||
else:
|
||||
result = pip.main(["install", plugin_package_name, "-i", mirror])
|
||||
success = result == 0
|
||||
if success:
|
||||
break
|
||||
else:
|
||||
nonebot.logger.warning(f"pip install failed, try next mirror.")
|
||||
except Exception as e:
|
||||
success = False
|
||||
continue
|
||||
|
||||
sys.stdout = sys.__stdout__
|
||||
sys.stderr = sys.__stderr__
|
||||
|
||||
return success, buffer.getvalue()
|
||||
|
||||
|
||||
def search_loaded_plugin(keyword: str) -> list[Plugin]:
|
||||
"""
|
||||
搜索已加载插件
|
||||
|
||||
Args:
|
||||
keyword (str): 关键词
|
||||
|
||||
Returns:
|
||||
list[Plugin]: 插件列表
|
||||
"""
|
||||
if nonebot.get_plugin(keyword) is not None:
|
||||
return [nonebot.get_plugin(keyword)]
|
||||
else:
|
||||
results = []
|
||||
for plugin in nonebot.get_loaded_plugins():
|
||||
if plugin.metadata is None:
|
||||
plugin.metadata = PluginMetadata(
|
||||
name=plugin.name, description="", usage=""
|
||||
)
|
||||
if (
|
||||
keyword
|
||||
in plugin.name + plugin.metadata.name + plugin.metadata.description
|
||||
):
|
||||
results.append(plugin)
|
||||
return results
|
||||
186
src/nonebot_plugins/liteyuki_pacman/rpm.py
Normal file
186
src/nonebot_plugins/liteyuki_pacman/rpm.py
Normal file
@ -0,0 +1,186 @@
|
||||
# 轻雪资源包管理器
|
||||
import os
|
||||
import zipfile
|
||||
import yaml
|
||||
from nonebot import require
|
||||
from nonebot.internal.matcher import Matcher
|
||||
from nonebot.permission import SUPERUSER
|
||||
|
||||
from src.utils.base.language import get_user_lang
|
||||
from src.utils.base.ly_typing import T_Bot, T_MessageEvent
|
||||
from src.utils.message.message import MarkdownMessage as md
|
||||
from src.utils.base.resource import (ResourceMetadata, add_resource_pack, change_priority, check_exist, check_status, get_loaded_resource_packs, get_resource_metadata, load_resources, remove_resource_pack)
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
from nonebot_plugin_alconna import Alconna, Args, on_alconna, Arparma, Subcommand
|
||||
|
||||
|
||||
@on_alconna(
|
||||
aliases={"资源包"},
|
||||
command=Alconna(
|
||||
"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
|
||||
).handle()
|
||||
async def _(bot: T_Bot, event: T_MessageEvent, result: Arparma, matcher: Matcher):
|
||||
ulang = get_user_lang(str(event.user_id))
|
||||
reply = ""
|
||||
send_as_md = False
|
||||
if result.subcommands.get("list"):
|
||||
send_as_md = True
|
||||
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()]
|
||||
# 遍历resources文件夹，获取未加载的资源包
|
||||
for folder in os.listdir("resources"):
|
||||
if folder not in loaded_folders:
|
||||
if 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 os.path.isfile(os.path.join("resources", folder)) and folder.endswith(".zip"):
|
||||
# zip文件
|
||||
# 临时解压并读取metadata.yml
|
||||
with zipfile.ZipFile(os.path.join("resources", folder), "r") as zip_ref:
|
||||
with zip_ref.open("metadata.yml") as f:
|
||||
metadata = ResourceMetadata(
|
||||
**yaml.load(f, Loader=yaml.FullLoader)
|
||||
)
|
||||
btn_load = md.btn_cmd(
|
||||
ulang.get("npm.install"),
|
||||
f"rpm load {folder}"
|
||||
)
|
||||
# 添加新行
|
||||
reply += (f"\n**{md.escape(metadata.name)}**({md.escape(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:
|
||||
pass
|
||||
if send_as_md:
|
||||
await md.send_md(reply, bot, event=event)
|
||||
else:
|
||||
await matcher.finish(reply)
|
||||
Reference in New Issue
Block a user