完成权限系统

2026-03-07 17:35:59 +08:00
parent e605527900
commit 230705f689
6 changed files with 232 additions and 25 deletions
--- a/konabot/plugins/perm_manage/init.py
+++ b/konabot/plugins/perm_manage/init.py
@ -0,0 +1,112 @@
+from typing import Annotated
+from nonebot.adapters import Event
+from nonebot.params import Depends
+from nonebot_plugin_alconna import Alconna, Args, Subcommand, UniMessage, on_alconna
+from konabot.common.pager import PagerQuery
+from konabot.common.permsys import DepPermManager, require_permission
+from konabot.common.permsys.entity import PermEntity, get_entity_chain_of_entity
+
+
+cmd = on_alconna(
+    Alconna(
+        "konaperm",
+        Subcommand(
+            "list",
+            Args["platform", str],
+            Args["entity_type", str],
+            Args["external_id", str],
+            Args["page?", int],
+        ),
+        Subcommand(
+            "get",
+            Args["platform", str],
+            Args["entity_type", str],
+            Args["external_id", str],
+            Args["perm", str],
+        ),
+        Subcommand(
+            "set",
+            Args["platform", str],
+            Args["entity_type", str],
+            Args["external_id", str],
+            Args["perm", str],
+            Args["val", str],
+        ),
+    ),
+    rule=require_permission("admin"),
+)
+
+
+async def _get_perm_entity_chain(platform: str, entity_type: str, external_id: str):
+    return get_entity_chain_of_entity(PermEntity(platform, entity_type, external_id))
+
+
+_DepEntityChain = Annotated[list[PermEntity], Depends(_get_perm_entity_chain)]
+
+
+def make_formatter(parent: PermEntity):
+    def _formatter(d: tuple[PermEntity, str, bool]):
+        permmark = {True: "[✅ ALLOW] ", False: "[❌ DENY] "}[d[2]]
+        inheritmark = ""
+        if parent != d[0]:
+            inheritmark = (
+                f"[继承自 {d[0].platform}.{d[0].entity_type}.{d[0].external_id}] "
+            )
+        return f"{permmark}{inheritmark}{d[1]}"
+
+    return _formatter
+
+
+@cmd.assign("list")
+async def list_permisison(
+    pm: DepPermManager,
+    ec: _DepEntityChain,
+    event: Event,
+    page: int = 1,
+):
+    pq = PagerQuery(page, 10)
+    data = await pm.list_permission(ec, pq)
+    msg = data.to_unimessage(make_formatter(ec[0]))
+    await msg.send(event)
+
+
+@cmd.assign("get")
+async def get_permisison(
+    pm: DepPermManager,
+    ec: _DepEntityChain,
+    perm: str,
+    event: Event,
+):
+    data = await pm.check_has_permission_info(ec, perm)
+
+    obj_s = f"{ec[0].platform}.{ec[0].entity_type}.{ec[0].external_id}"
+
+    if data is None:
+        await UniMessage.text(f"对象 {obj_s} 无 {perm} 权限记录").send(event)
+        return
+    pe, k, p = data
+    inheritmark = ""
+    if ec[0] != pe or k != perm:
+        inheritmark = (
+            f"继承自 {pe.platform}.{pe.entity_type}.{pe.external_id} 对 {k} 的设置，"
+        )
+    await UniMessage.text(f"{inheritmark}对象 {obj_s} 对 {perm} 的权限为 {p}").send(
+        event
+    )
+
+
+@cmd.assign("set")
+async def set_permisison(
+    pm: DepPermManager,
+    ec: _DepEntityChain,
+    perm: str,
+    val: str,
+    event: Event,
+):
+    if any(i == val.lower() for i in ("y", "yes", "allow", "true", "t")):
+        await pm.update_permission(ec[0], perm, True)
+    if any(i == val.lower() for i in ("n", "no", "deny", "false", "f")):
+        await pm.update_permission(ec[0], perm, False)
+    if any(i == val.lower() for i in ("null", "none")):
+        await pm.update_permission(ec[0], perm, None)
+    await get_permisison(pm, ec, perm, event)