添加一些权限目标

.drone.yml

@@ -39,7 +39,7 @@ steps:
     commands:
       - docker run --rm gitea.service.jazzwhom.top/mttu-developers/konabot:nightly-${DRONE_COMMIT_SHA} python scripts/test_plugin_load.py
       - docker run --rm gitea.service.jazzwhom.top/mttu-developers/konabot:nightly-${DRONE_COMMIT_SHA} python scripts/test_playwright.py
-      - docker run --rm gitea.service.jazzwhom.top/mttu-developers/konabot:nightly-${DRONE_COMMIT_SHA} python -m pytest --cov=./konabot/ --cov-report term-missing:skip-covered
+      - docker run -v ./tests:/app/tests:ro --rm gitea.service.jazzwhom.top/mttu-developers/konabot:nightly-${DRONE_COMMIT_SHA} python -m pytest --cov=./konabot/ --cov-report term-missing:skip-covered
   - name: 发送构建结果到 ntfy
     image: parrazam/drone-ntfy
     when:

konabot/plugins/bilibili_fetch/__init__.py

@@ -1,12 +1,14 @@
 import re
 
 from nonebot import get_plugin_config, on_message
+from nonebot.rule import Rule
 from nonebot_plugin_alconna import Reference, Reply, UniMsg
 
 from nonebot.adapters import Event
-from nonebot.adapters.onebot.v11.event import GroupMessageEvent as OB11GroupEvent
 from pydantic import BaseModel
 
+from konabot.common.permsys import require_permission
+
 
 class Config(BaseModel):
     bilifetch_enabled_groups: list[int] = []
@@ -19,11 +21,7 @@ pattern = (
 )
 
 
-def _rule(msg: UniMsg, evt: Event) -> bool:
+def _rule(msg: UniMsg) -> bool:
-    if isinstance(evt, OB11GroupEvent):
-        if evt.group_id not in config.bilifetch_enabled_groups:
-            return False
-
     to_search = msg.exclude(Reply, Reference).dump(json=True)
     to_search2 = msg.exclude(Reply, Reference).extract_plain_text()
     if not re.search(pattern, to_search) and not re.search(pattern, to_search2):
@@ -31,11 +29,11 @@ def _rule(msg: UniMsg, evt: Event) -> bool:
     return True
 
 
-matcher_fix = on_message(rule=_rule)
+matcher_fix = on_message(rule=Rule(_rule) & require_permission("bilifetch"))
+
 
 @matcher_fix.handle()
 async def _(event: Event):
     from nonebot_plugin_analysis_bilibili import handle_analysis
 
     await handle_analysis(event)
-

konabot/plugins/handle_text/__init__.py

@@ -70,8 +70,8 @@ async def _(msg: UniMsg, evt: Event, bot: Bot, target: DepLongTaskTarget):
         await target.send_message(res)
         return
 
-    env = TextHandlerEnvironment(is_trusted=False)
+    env = TextHandlerEnvironment(is_trusted=False, event=evt)
-    results = await runner.run_pipeline(res, istream or None, env)
+    results = await runner.run_pipeline(res, istream or None, evt, env)
 
     # 检查是否有错误
     for r in results:

konabot/plugins/handle_text/base.py

@@ -7,11 +7,13 @@ from string import whitespace
 from typing import cast
 
 from loguru import logger
+from nonebot.adapters import Event
 
 
 @dataclass
 class TextHandlerEnvironment:
     is_trusted: bool
+    event: Event | None = None
     buffers: dict[str, str] = field(default_factory=dict)
 
 
@@ -287,7 +289,7 @@ class PipelineRunner:
         env: TextHandlerEnvironment | None = None,
     ) -> list[TextHandleResult]:
         if env is None:
-            env = TextHandlerEnvironment(is_trusted=False, buffers={})
+            env = TextHandlerEnvironment(is_trusted=False, event=None, buffers={})
 
         results: list[TextHandleResult] = []
 

konabot/plugins/handle_text/handlers/ai_handlers.py

@@ -1,36 +1,51 @@
 from typing import Any, cast
 from konabot.common.llm import get_llm
-from konabot.plugins.handle_text.base import TextHandler, TextHandlerEnvironment, TextHandleResult
+from konabot.common.permsys import perm_manager
+from konabot.plugins.handle_text.base import (
+    TextHandler,
+    TextHandlerEnvironment,
+    TextHandleResult,
+)
 
 
 class THQwen(TextHandler):
     name = "qwen"
 
-    async def handle(self, env: TextHandlerEnvironment, istream: str | None, args: list[str]) -> TextHandleResult:
+    async def handle(
+        self, env: TextHandlerEnvironment, istream: str | None, args: list[str]
+    ) -> TextHandleResult:
+        pm = perm_manager()
+        if env.event is None or not pm.check_has_permission(env.event, "textfx.qwen"):
+            return TextHandleResult(
+                code=1,
+                ostream="这里暂未开启 AI 功能",
+            )
+
         llm = get_llm()
         messages = []
 
         if istream is not None:
-            messages.append({
+            messages.append({"role": "user", "content": istream})
-                "role": "user",
-                "content": istream
-            })
         if len(args) > 0:
-            message = ' '.join(args)
+            message = " ".join(args)
-            messages.append({
+            messages.append(
-                "role": "user",
+                {
-                "content": message,
+                    "role": "user",
-            })
+                    "content": message,
+                }
+            )
         if len(messages) == 0:
             return TextHandleResult(
                 code=1,
                 ostream="使用方法：qwen <提示词>",
             )
 
-        messages = [{
+        messages = [
-            "role": "system",
+            {
-            "content": "除非用户要求，请尽可能短点回答。另外，当前环境不支持 Markdown 语法，如果可以，请使用纯文本回答"
+                "role": "system",
-        }] + messages
+                "content": "除非用户要求，请尽可能短点回答。另外，当前环境不支持 Markdown 语法，如果可以，请使用纯文本回答",
+            }
+        ] + messages
         result = await llm.chat(cast(Any, messages))
         content = result.content
         if content is None:

konabot/plugins/mc_count_player/__init__.py

@@ -4,11 +4,16 @@ import mcstatus
 from nonebot import on_command
 from nonebot.adapters import Event
 from nonebot_plugin_alconna import UniMessage
-from konabot.common.nb.is_admin import is_admin
 from mcstatus.responses import JavaStatusResponse
 
+from konabot.common.permsys import require_permission
 
-cmd = on_command("宾几人", aliases=set(("宾人数", "mcbingo")), rule=is_admin)
+
+cmd = on_command(
+    "宾几人",
+    aliases=set(("宾人数", "mcbingo")),
+    rule=require_permission("minecraft.bingo.check"),
+)
 
 
 def parse_status(motd: str) -> str:
@@ -48,10 +53,11 @@ async def _(evt: Event):
         *map(lambda s: s[0].async_status(), servers),
         return_exceptions=True,
     )
-    messages = "\n".join((
+    messages = "\n".join(
-        dump_server_status(n, r)
+        (
-        for n, r in zip(map(lambda s: s[1], servers), responses)
+            dump_server_status(n, r)
-    ))
+            for n, r in zip(map(lambda s: s[1], servers), responses)
+        )
+    )
 
     await UniMessage.text(messages).finish(evt, at_sender=False)
-