zhenxun_bot/zhenxun/builtin_plugins/hooks/auth/auth_admin.py

import time

from nonebot_plugin_alconna import At
from nonebot_plugin_uninfo import Uninfo

from zhenxun.models.level_user import LevelUser
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.services.cache.runtime_cache import LevelUserMemoryCache, LevelUserSnapshot
from zhenxun.services.log import logger
from zhenxun.utils.utils import get_entity_ids

from .config import LOGGER_COMMAND, WARNING_THRESHOLD
from .exception import SkipPluginException
from .utils import send_message


async def auth_admin(
    plugin: PluginInfo,
    session: Uninfo,
    cached_levels: tuple[
        LevelUser | LevelUserSnapshot | None, LevelUser | LevelUserSnapshot | None
    ]
    | None = None,
):
    """管理员命令 个人权限

    参数:
        plugin: PluginInfo
        session: Uninfo
    """
    start_time = time.time()

    if not plugin.admin_level:
        return

    try:
        entity = get_entity_ids(session)

        global_user: LevelUser | LevelUserSnapshot | None = None
        group_users: LevelUser | LevelUserSnapshot | None = None

        if cached_levels is not None:
            global_user, group_users = cached_levels
        else:
            global_user, group_users = await LevelUserMemoryCache.get_levels(
                session.user.id, entity.group_id
            )

        user_level = global_user.user_level if global_user else 0
        if entity.group_id and group_users:
            user_level = max(user_level, group_users.user_level)

            if user_level < plugin.admin_level:
                await send_message(
                    session,
                    [
                        At(flag="user", target=session.user.id),
                        f"你的权限不足喔，该功能需要的权限等级: {plugin.admin_level}",
                    ],
                    entity.user_id,
                    background=True,
                )

                raise SkipPluginException(
                    f"{plugin.name}({plugin.module}) 管理员权限不足..."
                )
        elif global_user:
            if global_user.user_level < plugin.admin_level:
                await send_message(
                    session,
                    f"你的权限不足喔，该功能需要的权限等级: {plugin.admin_level}",
                    background=True,
                )

                raise SkipPluginException(
                    f"{plugin.name}({plugin.module}) 管理员权限不足..."
                )
    finally:
        # 记录执行时间
        elapsed = time.time() - start_time
        if elapsed > WARNING_THRESHOLD:  # 记录耗时超过500ms的检查
            logger.warning(
                f"auth_admin 耗时: {elapsed:.3f}s, plugin={plugin.module}",
                LOGGER_COMMAND,
                session=session,
            )