🐛 添加新异常判断跳过权限检测

zhenxun/builtin_plugins/chat_history/chat_message.py

@@ -2,13 +2,14 @@ from nonebot import on_message
 from nonebot.plugin import PluginMetadata
 from nonebot_plugin_alconna import UniMsg
 from nonebot_plugin_apscheduler import scheduler
-from nonebot_plugin_session import EventSession
+from nonebot_plugin_uninfo import Uninfo
 
 from zhenxun.configs.config import Config
 from zhenxun.configs.utils import PluginExtraData, RegisterConfig
 from zhenxun.models.chat_history import ChatHistory
 from zhenxun.services.log import logger
 from zhenxun.utils.enum import PluginType
+from zhenxun.utils.utils import get_entity_ids
 
 __plugin_meta__ = PluginMetadata(
     name="消息存储",
@@ -43,16 +44,15 @@ TEMP_LIST = []
 
 
 @chat_history.handle()
-async def _(message: UniMsg, session: EventSession):
+async def _(message: UniMsg, session: Uninfo):
-    # group_id = session.id3 or session.id2
+    entity = get_entity_ids(session)
-    group_id = session.id2
     TEMP_LIST.append(
         ChatHistory(
-            user_id=session.id1,
+            user_id=entity.user_id,
-            group_id=group_id,
+            group_id=entity.group_id,
             text=str(message),
             plain_text=message.extract_plain_text(),
-            bot_id=session.bot_id,
+            bot_id=session.self_id,
             platform=session.platform,
         )
     )

zhenxun/builtin_plugins/hooks/auth/auth_admin.py

@@ -22,11 +22,15 @@ async def auth_admin(plugin: PluginInfo, session: Uninfo):
         return
     entity = get_entity_ids(session)
     cache = Cache[list[LevelUser]](CacheType.LEVEL)
-    user_level = await cache.get(session.user.id) or []
+    user_list = await cache.get(session.user.id) or []
     if entity.group_id:
-        user_level += await cache.get(session.user.id, entity.group_id) or []
+        user_list += await cache.get(session.user.id, entity.group_id) or []
-        user = max(user_level, key=lambda x: x.user_level)
+        if user_list:
-        if user.user_level < plugin.admin_level:
+            user = max(user_list, key=lambda x: x.user_level)
+            user_level = user.user_level
+        else:
+            user_level = 0
+        if user_level < plugin.admin_level:
             await send_message(
                 session,
                 [
@@ -38,8 +42,8 @@ async def auth_admin(plugin: PluginInfo, session: Uninfo):
             raise SkipPluginException(
                 f"{plugin.name}({plugin.module}) 管理员权限不足..."
             )
-    elif user_level:
+    elif user_list:
-        user = max(user_level, key=lambda x: x.user_level)
+        user = max(user_list, key=lambda x: x.user_level)
         if user.user_level < plugin.admin_level:
             await send_message(
                 session,

zhenxun/builtin_plugins/hooks/auth/exception.py

@@ -12,3 +12,15 @@ class SkipPluginException(Exception):
 
     def __repr__(self) -> str:
         return self.info
+
+
+class PermissionExemption(Exception):
+    def __init__(self, info: str, *args: object) -> None:
+        super().__init__(*args)
+        self.info = info
+
+    def __str__(self) -> str:
+        return self.info
+
+    def __repr__(self) -> str:
+        return self.info

zhenxun/builtin_plugins/hooks/auth_checker.py

@@ -28,7 +28,11 @@ from .auth.auth_group import auth_group
 from .auth.auth_limit import LimitManage, auth_limit
 from .auth.auth_plugin import auth_plugin
 from .auth.config import LOGGER_COMMAND
-from .auth.exception import IsSuperuserException, SkipPluginException
+from .auth.exception import (
+    IsSuperuserException,
+    PermissionExemption,
+    SkipPluginException,
+)
 
 
 async def get_plugin_and_user(
@@ -41,10 +45,10 @@ async def get_plugin_and_user(
         user_id: 用户id
 
     异常:
-        SkipPluginException: 插件数据不存在
+        PermissionExemption: 插件数据不存在
-        SkipPluginException: 插件类型为HIDDEN
+        PermissionExemption: 插件类型为HIDDEN
-        SkipPluginException: 重复创建用户
+        PermissionExemption: 重复创建用户
-        SkipPluginException: 用户数据不存在
+        PermissionExemption: 用户数据不存在
 
     返回:
         tuple[PluginInfo, UserConsole]: 插件信息，用户信息
@@ -52,18 +56,18 @@ async def get_plugin_and_user(
     user_cache = Cache[UserConsole](CacheType.USERS)
     plugin = await Cache[PluginInfo](CacheType.PLUGINS).get(module)
     if not plugin:
-        raise SkipPluginException(f"插件:{module} 数据不存在，已跳过权限检查...")
+        raise PermissionExemption(f"插件:{module} 数据不存在，已跳过权限检查...")
     if plugin.plugin_type == PluginType.HIDDEN:
-        raise SkipPluginException(
+        raise PermissionExemption(
             f"插件: {plugin.name}:{plugin.module} 为HIDDEN，已跳过权限检查..."
         )
     user = None
     try:
         user = await user_cache.get(user_id)
     except IntegrityError as e:
-        raise SkipPluginException("重复创建用户，已跳过该次权限检查...") from e
+        raise PermissionExemption("重复创建用户，已跳过该次权限检查...") from e
     if not user:
-        raise SkipPluginException("用户数据不存在，已跳过权限检查...")
+        raise PermissionExemption("用户数据不存在，已跳过权限检查...")
     return plugin, user
 
 
@@ -143,7 +147,7 @@ async def auth(
     module = matcher.plugin_name or ""
     try:
         if not module:
-            raise SkipPluginException("Matcher插件名称不存在...")
+            raise PermissionExemption("Matcher插件名称不存在...")
         plugin, user = await get_plugin_and_user(module, entity.user_id)
         cost_gold = await get_plugin_cost(bot, user, plugin, session)
         await asyncio.gather(
@@ -162,6 +166,8 @@ async def auth(
         ignore_flag = True
     except IsSuperuserException:
         logger.debug("超级用户跳过权限检测...", LOGGER_COMMAND, session=session)
+    except PermissionExemption as e:
+        logger.info(str(e), LOGGER_COMMAND, session=session)
     if not ignore_flag and cost_gold > 0:
         await reduce_gold(entity.user_id, module, cost_gold, session)
     if ignore_flag: