Add files via upload

plugins/nonebot_plugin_manager/__init__.py

@@ -0,0 +1,110 @@
+from nonebot.plugin import on_shell_command, get_loaded_plugins, export
+from nonebot.matcher import Matcher
+from nonebot.typing import T_State
+from nonebot.exception import IgnoredException
+from nonebot.message import run_preprocessor
+from nonebot.adapters.cqhttp import Event, Bot, GroupMessageEvent, PrivateMessageEvent
+from configs.config import plugins2name_dict
+from models.ban_user import BanUser
+from .data import (
+    block_plugin,
+    unblock_plugin,
+    get_group_plugin_list,
+    auto_update_plugin_list,
+)
+from .parser import npm_parser
+
+# 导出给其他插件使用
+export = export()
+export.block_plugin = block_plugin
+export.unblock_plugin = unblock_plugin
+export.unblock_plugin = unblock_plugin
+export.get_group_plugin_list = get_group_plugin_list
+
+# 注册 shell_like 事件响应器
+plugin_manager = on_shell_command("npm", parser=npm_parser, priority=1)
+
+
+# 在 Matcher 运行前检测其是否启用
+@run_preprocessor
+async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
+    plugin = matcher.module
+    group_id = _get_group_id(event)
+    loaded_plugin_list = _get_loaded_plugin_list()
+    plugin_list = auto_update_plugin_list(loaded_plugin_list)
+
+    # 无视本插件的 Matcher
+    if plugin not in plugins2name_dict or matcher.priority in [1, 9] or await BanUser.isban(event.user_id):
+        return
+    try:
+        if isinstance(event, PrivateMessageEvent) and plugin_list[plugin]["default"]:
+            return
+    except:
+        pass
+
+    # print(matcher.module)
+    # print(f'plugin_list[plugin]["default"] = {plugin_list[plugin]["default"]}')
+    # print(f'{matcher.module} -> this is hook')
+    if not plugin_list[plugin]["default"]:
+        if event.message_type == 'group':
+            await bot.send_group_msg(group_id=event.group_id, message='此功能正在维护...')
+        else:
+            await bot.send_private_msg(user_id=event.user_id, message='此功能正在维护...')
+        raise IgnoredException(f"Nonebot Plugin Manager has blocked {plugin} !")
+    # print(plugin_list[plugin])
+    # print(group_id)
+    # print(plugin_list[plugin][group_id])
+    # print(not plugin_list[plugin][group_id])
+    if group_id in plugin_list[plugin]:
+        if not plugin_list[plugin][group_id]:
+            if plugin != 'ai' and matcher.type == 'message':
+                await bot.send_group_msg(group_id=event.group_id, message='该群未开启此功能..')
+            raise IgnoredException(f"Nonebot Plugin Manager has blocked {plugin} !")
+
+
+@plugin_manager.handle()
+async def _(bot: Bot, event: Event, state: T_State):
+    args = state["args"]
+    group_id = _get_group_id(event)
+    is_admin = _is_admin(event)
+    is_superuser = _is_superuser(bot, event)
+    if group_id == 0:
+        group_id = 'default'
+
+    if hasattr(args, "handle"):
+        await plugin_manager.finish(args.handle(args, group_id, is_admin, is_superuser))
+
+
+# 获取插件列表，并自动排除本插件
+def _get_loaded_plugin_list() -> list:
+    return list(
+        filter(
+            lambda plugin: plugin != "nonebot_plugin_manager",
+            map(lambda plugin: plugin.name, get_loaded_plugins()),
+        )
+    )
+
+
+def _get_group_id(event: Event) -> str:
+    try:
+        group_id = event.group_id
+    except AttributeError:
+        group_id = "default"
+    return str(group_id) if group_id else "default"
+
+
+def _is_admin(event: Event) -> bool:
+    return (
+        event.sender.role in ["admin", "owner"]
+        if isinstance(event, GroupMessageEvent)
+        else False
+    )
+
+
+def _is_superuser(bot: Bot, event: Event) -> bool:
+    return str(event.user_id) in bot.config.superusers
+
+
+plugins_dict = {
+
+}

plugins/nonebot_plugin_manager/data.py

@@ -0,0 +1,121 @@
+import json
+import httpx
+from pathlib import Path
+from configs.config import plugins2name_dict
+
+_DATA_PATH = Path() / "data" / "manager" / "plugin_list.json"
+
+
+def get_store_plugin_info(plugin: str) -> str:
+    store_plugin_list = _get_store_plugin_list()
+    if plugin in store_plugin_list:
+        plugin = store_plugin_list[plugin]
+        return (
+            f"ID: {plugin['id']}\n"
+            f"Name: {plugin['name']}\n"
+            f"Description: {plugin['desc']}\n"
+            f"Version: {httpx.get('https://pypi.org/pypi/'+plugin['link']+'/json').json()['info']['version']}\n"
+            f"Author: {plugin['author']}\n"
+            f"Repo: https://github.com/{plugin['repo']}"
+        )
+    else:
+        return "查无此插件！"
+
+
+def get_group_plugin_list(group_id: str) -> dict:
+    plugin_list = _load_plugin_list()
+    group_plugin_list = {}
+    for plugin in plugin_list:
+        if group_id in plugin_list[plugin]:
+            group_plugin_list[plugin] = plugin_list[plugin][group_id]
+        else:
+            group_plugin_list[plugin] = plugin_list[plugin]["default"]
+    return group_plugin_list
+
+
+def get_store_pulgin_list() -> str:
+    message = "商店插件列表如下："
+    for plugin in _get_store_plugin_list():
+        if plugin in _load_plugin_list() or plugin == "nonebot_plugin_manager":
+            message += f"\n[o] {plugin}"
+        else:
+            message += f"\n[x] {plugin}"
+    return message
+
+
+def auto_update_plugin_list(loaded_plugin_list: list, keep_history: bool = False):
+    plugin_list = _load_plugin_list()
+    for plugin in loaded_plugin_list:
+        if plugin not in plugin_list:
+            plugin_list[plugin] = {"default": True}
+    if not keep_history:
+        plugin_list = {
+            key: plugin_list[key] for key in plugin_list if key in loaded_plugin_list
+        }
+    _dump_plugin_list(plugin_list)
+    return plugin_list
+
+
+def block_plugin(group_id: str, *plugins: str):
+    return _update_plugin_list(group_id, True, *plugins)
+
+
+def unblock_plugin(group_id: str, *plugins: str):
+    return _update_plugin_list(group_id, False, *plugins)
+
+
+# 获取商店插件列表
+def _get_store_plugin_list() -> dict:
+    store_plugin_list = {}
+    for plugin in httpx.get(
+        "https://cdn.jsdelivr.net/gh/nonebot/nonebot2@master/docs/.vuepress/public/plugins.json"
+    ).json():
+        store_plugin_list.update({plugin["id"]: plugin})
+    return store_plugin_list
+
+
+# 更新插件列表
+def _update_plugin_list(group_id: str, block: bool, *plugins: str) -> str:
+    plugin_list = _load_plugin_list()
+    message = "结果如下："
+    operate = "屏蔽" if block else "启用"
+    for plugin in plugins:
+        for values in plugins2name_dict.values():
+            if plugin in values:
+                plugin = list(plugins2name_dict.keys())[list(plugins2name_dict.values()).index(values)]
+                # print(plugin)
+                break
+        message += "\n"
+        if plugin in plugin_list:
+            if (
+                not group_id in plugin_list[plugin]
+                or plugin_list[plugin][group_id] == block
+            ):
+                plugin_list[plugin][group_id] = not block
+                message += f"插件{plugin}{operate}成功！"
+                print(plugin_list[plugin][group_id])
+            else:
+                message += f"插件{plugin}已经{operate}！"
+        else:
+            message += f"插件{plugin}不存在！"
+    _dump_plugin_list(plugin_list)
+    return message
+
+
+# 加载插件列表
+def _load_plugin_list() -> dict:
+    try:
+        return json.load(_DATA_PATH.open("r", encoding="utf-8"))
+    except FileNotFoundError:
+        return {}
+
+
+# 保存插件列表
+def _dump_plugin_list(plugin_list: dict):
+    _DATA_PATH.parent.mkdir(parents=True, exist_ok=True)
+    json.dump(
+        plugin_list,
+        _DATA_PATH.open("w", encoding="utf-8"),
+        indent=4,
+        separators=(",", ": "),
+    )

plugins/nonebot_plugin_manager/handle.py

@@ -0,0 +1,138 @@
+from argparse import Namespace
+
+from .data import *
+
+
+def handle_list(
+    args: Namespace,
+    group_id: str,
+    is_admin: bool,
+    is_superuser: bool,
+) -> str:
+    message = ""
+
+    if args.store:
+        if is_superuser:
+            return get_store_pulgin_list()
+        else:
+            return "获取商店插件列表需要超级用户权限！"
+
+    if args.default:
+        if is_superuser:
+            group_id = "default"
+            message += "默认"
+        else:
+            return "获取默认插件列表需要超级用户权限！"
+
+    if args.group:
+        if is_superuser:
+            group_id = args.group
+            message += f"群{args.group}"
+        else:
+            return "获取指定群插件列表需要超级用户权限！"
+
+    message += "插件列表如下：\n"
+    message += "\n".join(
+        f"[{'o' if get_group_plugin_list(group_id)[plugin] else 'x'}] {plugin}"
+        for plugin in get_group_plugin_list(group_id)
+    )
+    return message
+
+
+def handle_block(
+    args: Namespace,
+    group_id: str,
+    is_admin: bool,
+    is_superuser: bool,
+) -> str:
+
+    if not is_admin and not is_superuser:
+        return "管理插件需要群管理员权限！"
+
+    message = ""
+
+    if args.default:
+        if is_superuser:
+            group_id = "default"
+            message += "默认"
+        else:
+            return "管理默认插件需要超级用户权限！"
+
+    if args.group:
+        if is_superuser:
+            group_id = args.group
+            message += f"群{args.group}"
+        else:
+            return "管理指定群插件需要超级用户权限！"
+
+    message += block_plugin(group_id, *args.plugins)
+    return message
+
+
+def handle_unblock(
+    args: Namespace,
+    group_id: str,
+    is_admin: bool,
+    is_superuser: bool,
+) -> str:
+    message = ""
+
+    if not is_admin and not is_superuser:
+        return "管理插件需要群管理员权限！"
+
+    if args.default:
+        if is_superuser:
+            group_id = "default"
+            message += "默认"
+        else:
+            return "管理默认插件需要超级用户权限！"
+
+    if args.group:
+        if is_superuser:
+            group_id = args.group
+            message += f"群{args.group}"
+        else:
+            return "管理指定群插件需要超级用户权限！"
+
+    message += unblock_plugin(group_id, *args.plugins)
+    return message
+
+
+def handle_info(
+    args: Namespace,
+    group_id: str,
+    is_admin: bool,
+    is_superuser: bool,
+) -> str:
+
+    if not is_admin and not is_superuser:
+        return "管理插件需要超级权限！"
+
+    return get_store_plugin_info(args.plugin)
+
+
+def handle_install(
+    args: Namespace,
+    group_id: str,
+    is_admin: bool,
+    is_superuser: bool,
+) -> str:
+    pass
+
+
+def handle_update(
+    args: Namespace,
+    group_id: str,
+    is_admin: bool,
+    is_superuser: bool,
+) -> str:
+    pass
+
+
+def handle_uninstall(
+    args: Namespace,
+    group_id: str,
+    is_admin: bool,
+    is_superuser: bool,
+) -> str:
+    pass

plugins/nonebot_plugin_manager/parser.py

@@ -0,0 +1,60 @@
+from nonebot.rule import ArgumentParser
+
+from .handle import *
+
+npm_parser = ArgumentParser("npm", add_help=False)
+npm_parser.add_argument(
+    "-h", "--help", action="store_true", help="show this help message and exit"
+)
+
+npm_subparsers = npm_parser.add_subparsers()
+
+list_parser = npm_subparsers.add_parser("list", help="show plugin list")
+list_parser.add_argument(
+    "-s", "--store", action="store_true", help="show plugin store list"
+)
+list_parser.add_argument(
+    "-d", "--default", action="store_true", help="show default plugin list"
+)
+list_parser.add_argument("-g", "--group", action="store", help="show group plugin list")
+list_parser.set_defaults(handle=handle_list)
+
+block_parser = npm_subparsers.add_parser("block", help="block plugin")
+block_parser.add_argument("plugins", nargs="*", help="plugins you want to block")
+block_parser.add_argument("-d", "--default", action="store_true", help="set default")
+block_parser.add_argument("-a", "--all", action="store_true", help="select all plugin")
+block_parser.add_argument("-g", "--group", action="store", help="set in group")
+block_parser.set_defaults(handle=handle_block)
+
+unblock_parser = npm_subparsers.add_parser("unblock", help="unblock plugin")
+unblock_parser.add_argument("plugins", nargs="*", help="plugins you want to unblock")
+unblock_parser.add_argument("-d", "--default", action="store_true", help="set default")
+unblock_parser.add_argument(
+    "-a", "--all", action="store_true", help="select all plugin"
+)
+unblock_parser.add_argument("-g", "--group", action="store", help="set in group")
+unblock_parser.set_defaults(handle=handle_unblock)
+
+info_parser = npm_subparsers.add_parser("info", help="show plugin info")
+info_parser.add_argument("plugin", help="plugins you want to know")
+info_parser.set_defaults(handle=handle_info)
+
+install_parser = npm_subparsers.add_parser("install", help="install plugin")
+install_parser.add_argument("plugins", nargs="*", help="plugins you want to install")
+install_parser.add_argument("-i", "--index", action="store", help="point to a mirror")
+install_parser.set_defaults(handle=handle_install)
+
+update_parser = npm_subparsers.add_parser("update", help="update plugin")
+update_parser.add_argument("plugins", nargs="*", help="plugins you want to update")
+update_parser.add_argument("-a", "--all", action="store_true", help="select all plugin")
+update_parser.add_argument("-i", "--index", action="store", help="point to a mirror")
+update_parser.set_defaults(handle=handle_update)
+
+uninstall_parser = npm_subparsers.add_parser("uninstall", help="uninstall plugin")
+uninstall_parser.add_argument(
+    "plugins", nargs="*", help="plugins you want to uninstall"
+)
+uninstall_parser.add_argument(
+    "-a", "--all", action="store_true", help="select all plugin"
+)
+uninstall_parser.set_defaults(handle=handle_uninstall)