更新 SayaManager，支持不可重载/卸载的metadata

Author: SAGIRI-kawaii

sagiri_bot/config.py

@@ -5,7 +5,7 @@
 from pathlib import Path
 from pydantic import BaseModel
 from typing_extensions import TypedDict
-from typing import Type, List, Dict, Union
+from typing import Type, List, Dict, Union, Any
 
 from creart import add_creator
 from creart import exists_module
@@ -23,6 +23,7 @@ class PluginMeta(BaseModel):
     icon: str = ""
     prefix: List[str] = []
     triggers: List[str] = []
+    metadata: Dict[str, Any] = {}
 
 
 def load_plugin_meta(path: Union[Path, str]) -> PluginMeta:

sagiri_bot/handler/required_module/bot_management/metadata.json

@@ -9,5 +9,9 @@
   "usage": "发送 setting -set key1=value1 key2=value2 ... 改变群内设置\n发送 user -grant @target [1-3] 改变成员权限等级\n发送 blacklist -add @target 添加群内黑名单\n发送 blacklist -remove @target 移除群内黑名单",
   "icon": "",
   "prefix": [],
-  "triggers": []
+  "triggers": [],
+  "metadata": {
+    "uninstallable": false,
+    "reloadable": false
+  }
 }

sagiri_bot/handler/required_module/chat_recorder_handler/metadata.json

@@ -9,5 +9,9 @@
   "usage": "自动触发",
   "icon": "",
   "prefix": [],
-  "triggers": []
+  "triggers": [],
+  "metadata": {
+    "uninstallable": false,
+    "reloadable": false
+  }
 }

sagiri_bot/handler/required_module/exception_catcher/metadata.json

@@ -9,5 +9,9 @@
   "usage": "后台发生错误时自动触发",
   "icon": "",
   "prefix": [],
-  "triggers": []
+  "triggers": [],
+  "metadata": {
+    "uninstallable": false,
+    "reloadable": true
+  }
 }

sagiri_bot/handler/required_module/helper/metadata.json

@@ -9,5 +9,9 @@
   "usage": "发送 `/help` 即可查看所有插件",
   "icon": "",
   "prefix": ["/", ".", "#", ""],
-  "triggers": ["help", "帮助", "菜单", "功能", "纱雾帮助"]
+  "triggers": ["help", "帮助", "菜单", "功能", "纱雾帮助"],
+  "metadata": {
+    "uninstallable": false,
+    "reloadable": true
+  }
 }

sagiri_bot/handler/required_module/message_statistics/metadata.json

@@ -9,5 +9,9 @@
   "usage": "在群中发送 `消息量统计` 即可\n发送 `消息量统计 -group` 可查看本群统计",
   "icon": "",
   "prefix": ["", "/"],
-  "triggers": ["消息量统计"]
+  "triggers": ["消息量统计"],
+  "metadata": {
+    "uninstallable": false,
+    "reloadable": true
+  }
 }

sagiri_bot/handler/required_module/mirai_event/metadata.json

@@ -9,5 +9,9 @@
   "usage": "自动触发",
   "icon": "",
   "prefix": [],
-  "triggers": []
+  "triggers": [],
+  "metadata": {
+    "uninstallable": false,
+    "reloadable": true
+  }
 }

sagiri_bot/handler/required_module/saya_manager/__init__.py

@@ -1,7 +1,8 @@
 import os
 import re
-from typing import List, Dict, Optional, Union
+from typing import List, Dict, Union
 
+from creart import create
 from graia.saya import Saya, Channel
 from graia.ariadne.app import Ariadne
 from graia.broadcast.interrupt.waiter import Waiter
@@ -11,11 +12,11 @@
 from graia.saya.builtins.broadcast.schema import ListenerSchema
 from graia.ariadne.event.message import Group, Member, GroupMessage
 
-from .utils import saya_data
+from .utils import saya_data, uninstallable, reloadable
 from sagiri_bot.internal_utils import MessageChainUtils
 from sagiri_bot.internal_utils import user_permission_require
 
-saya = Saya.current()
+saya = create(Saya)
 channel = Channel.current()
 
 channel.name("SayaManager")
@@ -30,10 +31,6 @@
 inc = InterruptControl(saya.broadcast)
 
 
-def get_loaded_channels() -> Dict[str, Channel]:
-    return saya.channels
-
-
 def get_all_channels() -> List[str]:
     ignore = ["__init__.py", "__pycache__"]
     dirs = [
@@ -54,13 +51,9 @@ def get_all_channels() -> List[str]:
 
 
 def get_unloaded_channels() -> List[str]:
-    loaded_channels = get_loaded_channels()
+    loaded_channels = saya.channels
     all_channels = get_all_channels()
-    return [channel for channel in all_channels if channel not in loaded_channels]
-
-
-def get_channel(name: str) -> Optional[Channel]:
-    return get_loaded_channels().get(name)
+    return [c for c in all_channels if c not in loaded_channels]
 
 
 def load_channel(modules: Union[str, List[str]]) -> Dict[str, Exception]:
@@ -83,37 +76,37 @@ def unload_channel(modules: Union[str, List[str]]) -> Dict[str, Exception]:
     exceptions = {}
     if isinstance(modules, str):
         modules = [modules]
-    loaded_channels = get_loaded_channels()
+    loaded_channels = saya.channels
     channels_to_unload = {
         module: loaded_channels[module]
         for module in modules
         if module in loaded_channels
     }
     with saya.module_context():
-        for channel, value in channels_to_unload.items():
+        for c, value in channels_to_unload.items():
             try:
                 saya.uninstall_channel(value)
             except Exception as e:
-                exceptions[channel] = e
+                exceptions[c] = e
     return exceptions
 
 
 def reload_channel(modules: Union[str, List[str]]) -> Dict[str, Exception]:
     exceptions = {}
     if isinstance(modules, str):
         modules = [modules]
-    loaded_channels = get_loaded_channels()
+    loaded_channels = saya.channels
     channels_to_reload = {
         module: loaded_channels[module]
         for module in modules
         if module in loaded_channels
     }
     with saya.module_context():
-        for channel, value in channels_to_reload.items():
+        for c, value in channels_to_reload.items():
             try:
                 saya.reload_channel(value)
             except Exception as e:
-                exceptions[channel] = e
+                exceptions[c] = e
     return exceptions
 
 
@@ -122,7 +115,7 @@ async def saya_manager(
     app: Ariadne, message: MessageChain, group: Group, member: Member, source: Source
 ):
     if message.display.strip() == "已加载插件":
-        loaded_channels = get_loaded_channels()
+        loaded_channels = saya.channels
         keys = list(loaded_channels.keys())
         keys.sort()
         return await app.send_group_message(
@@ -143,20 +136,20 @@ async def saya_manager(
         )
     elif re.match(r"插件详情 .+", message.display):
         target = message.display[5:].strip()
-        loaded_channels = get_loaded_channels()
+        loaded_channels = saya.channels
         keys = list(loaded_channels.keys())
         if target.isdigit():
             keys.sort()
             if not 0 <= int(target) - 1 < len(keys):
                 return await app.send_group_message(
                     group, MessageChain("错误的编号！请检查后再发送！"), quote=source
                 )
-            channel = loaded_channels[keys[int(target) - 1]]
+            c = loaded_channels[keys[int(target) - 1]]
             channel_path = keys[int(target) - 1]
         else:
             for lchannel in loaded_channels.keys():
-                if loaded_channels[lchannel]._name == target:
-                    channel = loaded_channels[lchannel]
+                if loaded_channels[lchannel].meta["name"] == target:
+                    c = loaded_channels[lchannel]
                     channel_path = lchannel
                     break
             else:
@@ -167,9 +160,9 @@ async def saya_manager(
             group,
             MessageChain(
                 [
-                    Plain(text=f"插件名称：{channel._name}\n"),
-                    Plain(text=f"插件作者：{'、'.join(channel._author)}\n"),
-                    Plain(text=f"插件描述：{channel._description}\n"),
+                    Plain(text=f"插件名称：{c.meta['name']}\n"),
+                    Plain(text=f"插件作者：{'、'.join(c.meta['author'])}\n"),
+                    Plain(text=f"插件描述：{c.meta['description']}\n"),
                     Plain(text=f"插件包名：{channel_path}"),
                 ]
             ),
@@ -207,18 +200,18 @@ async def saya_manager(
                 return await app.send_group_message(
                     group, MessageChain("错误的编号！请检查后再发送！"), quote=source
                 )
-            channel = unloaded_channels[int(target) - 1]
+            c = unloaded_channels[int(target) - 1]
         else:
             for ulchannel in unloaded_channels:
                 if ulchannel == target:
-                    channel = ulchannel
+                    c = ulchannel
                     break
             else:
                 return await app.send_group_message(
                     group, MessageChain("错误的名称！请检查后再发送！"), quote=source
                 )
 
-        await app.send_message(group, MessageChain(f"你确定要加载插件 `{channel}` 吗？（是/否）"))
+        await app.send_message(group, MessageChain(f"你确定要加载插件 `{c}` 吗？（是/否）"))
 
         @Waiter.create_using_function([GroupMessage])
         def confirm_waiter(
@@ -236,10 +229,10 @@ def confirm_waiter(
                 group, MessageChain("非预期回复，进程退出"), quote=source
             )
         elif result == "是":
-            result = load_channel(channel)
+            result = load_channel(c)
             if result:
                 return await app.send_group_message(
-                    group, MessageChain(f"发生错误：{result[channel]}"), quote=source
+                    group, MessageChain(f"发生错误：{result[c]}"), quote=source
                 )
             else:
                 return await app.send_group_message(
@@ -256,30 +249,37 @@ def confirm_waiter(
             )
         load_type = "reload" if message.display[0] == "重" else "unload"
         target = message.display[5:].strip()
-        loaded_channels = get_loaded_channels()
+        loaded_channels = saya.channels
         keys = list(loaded_channels.keys())
         keys.sort()
         if target.isdigit():
             if not 0 <= int(target) - 1 < len(keys):
                 return await app.send_group_message(
                     group, MessageChain("错误的编号！请检查后再发送！"), quote=source
                 )
-            channel = loaded_channels[keys[int(target) - 1]]
+            c = loaded_channels[keys[int(target) - 1]]
             channel_path = keys[int(target) - 1]
         else:
             for lchannel in loaded_channels.keys():
-                if loaded_channels[lchannel]._name == target:
-                    channel = loaded_channels[lchannel]
+                if loaded_channels[lchannel].meta["name"] == target:
+                    c = loaded_channels[lchannel]
                     channel_path = lchannel
                     break
             else:
                 return await app.send_group_message(
                     group, MessageChain("错误的名称！请检查后再发送！"), quote=source
                 )
-
+        if load_type == "reload" and not reloadable(c.module):
+            return await app.send_group_message(
+                group, MessageChain(f"插件 `{c.meta['name']}` 不可重载！"), quote=source
+            )
+        if load_type == "unload" and not uninstallable(c.module):
+            return await app.send_group_message(
+                group, MessageChain(f"插件 `{c.meta['name']}` 不可卸载！"), quote=source
+            )
         await app.send_message(
             group,
-            MessageChain(f"你确定要{message.display[0]}载插件 `{channel._name}` 吗？（是/否）"),
+            MessageChain(f"你确定要{message.display[0]}载插件 `{c.meta['name']}` 吗？（是/否）"),
         )
 
         @Waiter.create_using_function([GroupMessage])
@@ -322,7 +322,7 @@ def confirm_waiter(
             )
         switch_type = "on" if message.display[:2] == "打开" else "off"
         target = message.display[5:].strip()
-        loaded_channels = get_loaded_channels()
+        loaded_channels = saya.channels
         keys = list(loaded_channels.keys())
         keys.sort()
         channel_path = ""
@@ -334,7 +334,7 @@ def confirm_waiter(
             channel_path = keys[int(target) - 1]
         else:
             for lchannel in loaded_channels.keys():
-                if loaded_channels[lchannel]._name == target:
+                if loaded_channels[lchannel].meta["name"] == target:
                     channel_path = lchannel
                     break
         saya_data.switch_on(

sagiri_bot/handler/required_module/saya_manager/metadata.json

@@ -9,5 +9,9 @@
   "usage": "发送 `已加载插件` 查看已加载插件\n发送 `插件详情 [编号|名称]` 可查看插件详情\n发送 `[加载|重载|卸载|打开|关闭]插件 [编号|名称]` 可加载/重载/卸载/打开/关闭插件",
   "icon": "",
   "prefix": [],
-  "triggers": []
+  "triggers": [],
+  "metadata": {
+    "uninstallable": false,
+    "reloadable": false
+  }
 }

sagiri_bot/handler/required_module/saya_manager/utils.py

@@ -15,6 +15,8 @@
 from graia.ariadne.event.mirai import MiraiEvent, GroupEvent
 from graia.saya.builtins.broadcast.schema import ListenerSchema
 
+from sagiri_bot.config import load_plugin_meta_by_module
+
 DEFAULT_SWITCH = True
 DEFAULT_NOTICE = False
 
@@ -70,7 +72,8 @@ def saya_init():
             if isinstance(cube.metaclass, ListenerSchema):
                 bcc.removeListener(bcc.getListener(cube.content))
                 if all(
-                    [issubclass(i, GroupEvent) for i in cube.metaclass.listening_events]
+                    issubclass(i, GroupEvent)
+                    for i in cube.metaclass.listening_events
                 ):
                     cube.metaclass.decorators.append(manageable(channel.module))
                 else:
@@ -241,4 +244,14 @@ def load(
         return self
 
 
+def reloadable(module: str) -> bool:
+    plugin_meta = load_plugin_meta_by_module(module)
+    return plugin_meta.metadata.get("reloadable", True)
+
+
+def uninstallable(module: str) -> bool:
+    plugin_meta = load_plugin_meta_by_module(module)
+    return plugin_meta.metadata.get("uninstallable", True)
+
+
 saya_data = SayaData().load()

sagiri_bot/handler/required_module/system_status/metadata.json

@@ -9,5 +9,9 @@
   "usage": "发送 /sys 或 '/sys -a' 或 '/sys -all' 查看CPU、内存以及图库占用信息\n发送 /sys -i 或 '/sys -info' 查看CPU、内存信息\n发送 /sys -s 或 '/sys -storage' 查看图库占用信息",
   "icon": "",
   "prefix": ["/"],
-  "triggers": ["sys", "system"]
+  "triggers": ["sys", "system"],
+  "metadata": {
+    "uninstallable": false,
+    "reloadable": true
+  }
 }