import os import sys import aiohttp import nonebot.plugin import pip from io import StringIO from arclet.alconna import MultiVar from nonebot import require from nonebot.exception import FinishedException, IgnoredException 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 from liteyuki.utils.data_manager import InstalledPlugin from liteyuki.utils.language import get_user_lang from liteyuki.utils.ly_typing import T_Bot from liteyuki.utils.message import Markdown as md from liteyuki.utils.permission import GROUP_ADMIN, GROUP_OWNER from liteyuki.utils.tools import clamp from .common import * require("nonebot_plugin_alconna") from nonebot_plugin_alconna import on_alconna, Alconna, Args, Arparma, Subcommand # 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], alias=["启用"], ), Subcommand( "disable", Args["plugin_name", str], alias=["停用"], ), Subcommand( enable_global, Args["plugin_name", str], alias=["全局启用"], ), Subcommand( disable_global, Args["plugin_name", str], alias=["全局停用"], ), # 安装部分 Subcommand( "update", alias=["u"], ), Subcommand( "search", Args["keywords", MultiVar(str)]["show_num", int, 15], 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", "列表"], ), Subcommand( "usage", Args["plugin_name", str], alias=["详情"], ) ) ).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 result.subcommands.get("disable"): toggle = result.subcommands.get("enable") is not None plugin_exist = get_plugin_exist(plugin_name) session_enable = get_plugin_session_enable(event, plugin_name) # 获取插件当前状态 default_enable = get_plugin_default_enable(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"))) 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_name) else: session.enabled_plugins.append(plugin_name) else: if default_enable: session.disabled_plugins.append(plugin_name) else: session.enabled_plugins.remove(plugin_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_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(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: storePlugin = plugin_db.first(GlobalPlugin(), "module_name = ?", plugin_name, default=GlobalPlugin(module_name=plugin_name)) if toggle: storePlugin.enabled = True else: storePlugin.enabled = False plugin_db.upsert(storePlugin) except Exception as e: print(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 = result.subcommands.get("search").args.get("show_num") 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 = 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.first(InstalledPlugin(), "module_name = ?", plugin_name) # 查询数据库中是否已经安装 if r_load: if found_in_db_plugin is None: plugin_db.upsert(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") found_installed_plugin: InstalledPlugin = plugin_db.first(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"npm usage {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} {btn_homepage}" 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.first(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" await md.send_md(reply, bot, event=event) elif sc.get("usage"): # TODO pass else: pass @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") 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]: 插件列表 """ 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: 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 def npm_install(plugin_package_name) -> tuple[bool, str]: """ 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"npm_install failed, try next mirror.") except Exception as e: success = False continue sys.stdout = sys.__stdout__ sys.stderr = sys.__stderr__ return success, buffer.getvalue()