拦截器

主要学习事件拦截。stop_propagation、hook、ALC

话说开始写之后，发现有些同类的：

• 黑名单插件 nonebot-plugin-authrespond
  • 原理为插件运行前监听Matcher响应器属性
  • 全平台支持,但是 二级群组可能存在问题
• 权限控制插件 nonebot-plugin-access-control
  • 有频率控制
  • 需要数据库

下面的示例仅供参考学习，再深入的话参考上面的插件源码

V1 stop_propagation方案，但可能拦截失败

from nonebot import get_plugin_config
from nonebot.plugin import PluginMetadata

from .config import Config

__plugin_meta__ = PluginMetadata(
    name="blocker",
    description="拦截器/别名器。最高优先级，静态/动态拦截插件 (相当于开关插件) 以及增加别名",
    usage="暂静态生效，等待开发动态功能",
    config=Config,
)

config = get_plugin_config(Config)

# 一开始用的stop_propagation方案。缺点：默认优先级好像是1，同优先级无法拦截。社区插件和内置插件基本不可拦截
# 所以后面弃用了，换用钩子函数进行拦截

from nonebot import get_driver, logger
from nonebot import on_message
from nonebot.adapters.onebot.v11 import Bot, MessageEvent
from nonebot.matcher import Matcher

superusers = set(map(str, get_driver().config.superusers)) # 超级用户列表

m = on_message(priority=1, block=False)

@m.handle()
async def fn(event: MessageEvent, matcher: Matcher):
    logger.info(f'superusers:{superusers}, user_id:{event.user_id}, isBlock:{str(event.user_id) not in superusers}')
    if (str(event.user_id) not in superusers):
        matcher.stop_propagation()

V2 钩子方案，但不通用。无法拦截非onebot

event_preprocessor 钩子

from nonebot import get_plugin_config
from nonebot.plugin import PluginMetadata

from .config import Config

__plugin_meta__ = PluginMetadata(
    name="blocker",
    description="拦截器/别名器。最高优先级，静态/动态拦截插件 (相当于开关插件) 以及增加别名",
    usage="暂静态生效，等待开发动态功能",
    config=Config,
)

config = get_plugin_config(Config)

from nonebot import get_driver, logger
from nonebot.exception import IgnoredException
from nonebot.message import event_preprocessor
from nonebot.adapters import Event
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent

superusers = set(map(str, get_driver().config.superusers)) # 超级用户列表


@event_preprocessor
async def blocker(event: Event):
    # 拦截所有非艾特自己的信息
    # if not event.is_tome(): raise IgnoredException("some reason1")

    # 拦截所有非超管发言
    # if hasattr(event, 'user_id'):
    user_id = getattr(event, 'user_id', None)
    if user_id is not None and str(user_id) not in superusers:
        # logger.info(f'拦截, {user_id}, {event.user_id}, {event.get_user_id}')
        raise IgnoredException("some reason2")

# import os

# 处理gpt插件的上下文，将群昵称注入deepseek__prompt中
@event_preprocessor
async def group_env(event: GroupMessageEvent, event2: Event, bot: Bot):
    if '/ds' in event.raw_message:
        group_info = await bot.get_group_info(group_id=event.group_id)
        group_name = group_info.get("group_name", "未知群组")
        logger.info(f'event, {event}, {event.group_id}, {group_info}, {group_name}')
        # os.environ["deepseek__prompt"] = f'这是群组"${group_name}"中的提问，请用中文简短回答' # 方法一。无用，不会更新已缓存的配置值
        # get_driver().config.get("deepseek__prompt", None) = f'这是群组"${group_name}"中的提问，请用中文简短回答' # 方法二。报错，表达式不能是赋值目标
        # get_driver().config.deepseek__prompt = f'这是群组"${group_name}"中的提问，请用中文简短回答' # 方法二-2。报错，无法为类型“Config”分配成员“deepseek__prompt”
        # config = get_driver().config; config.deepseek.prompt = f'这是群组"{group_name}"中的提问，请用中文简短回答' # 方法二-3。报错，'dict' object has no attribute 'prompt'
        # event.raw_message = f'prompt: 这是群组"group_name"中的提问，请用中文简短回答\n' + event.raw_message # 方法三。对于我写的其他onebot插件是对的，但对于非onebot会有问题，不通用
    event.raw_message = event.raw_message + '测试' # 仅对我写的其他onebot插件正确，对其他插件不成功

V3 再加上ALC + 重新抛出的思路

这里可以详见我的仓库中的 nonebot-linc 库的blocker插件代码

@run_preprocessor
async def _(matcher: Matcher, bot: Bot, state: T_State,
    event: Event, event_qq: GroupMessageEvent, msg: UniMsg
):
    if '/debug4' in str(event.get_message()):
        logger.info(f'''
            msg_event: {event.get_message()}
            msg_qq: {event_qq.raw_message}
            msg_uni: {msg}
        ''')
        # event.__setattr__("message", "new_message_event") # 会有bug，message不是字符串
        event.get_message()[0] = '/debug3 测试'
        event_qq.raw_message = "new_message_qq"
        msg.clear(); msg.append('new_message_uni')
        logger.info(f'''
            msg_event2: {event.get_message()}
            msg_qq2: {event_qq.raw_message}
            msg_uni2: {msg}
        ''') 

        fake_event = deepcopy(event)
        # fake_event.__setattr__("message", "/echo new_message_event222")
        # setattr(fake_event, "_is_fake", True)
        # await handle_event(bot, fake_event)
        raise IgnoredException("some reason")
    
    if '/ds' in str(event.get_message()):
        if hasattr(event, "_is_fake"): return
        del event.get_message()[0]; event.get_message().append('/ds 你今年多大了') # 不生效
        msg.clear(); msg.append('/ds 今天是什么日期') # 生效
        event_qq.raw_message = "/ds 你叫什么名字" # 不生效 (仅于其他onebot插件才生效)
        logger.info(f'''
            /ds blocker
            msg_event: {event.get_message()}            
            msg_uni: {msg}
            msg_qq: {event_qq.raw_message}
        ''')
        fake_event = deepcopy(event)
        setattr(fake_event, "_is_fake", True)
        await handle_event(bot, fake_event) # 修改的是副本，必须拦截并重新抛出事件
        raise IgnoredException("some reason")

nonebot-plugin-authrespond

.github
nonebot_lugin_authrespond
__init__.py	入口
config.py
cubp.py
perm.py	核心
plugins_data.py
run.py	核心，钩子
.gitignore
LICENSE
README.md
pyproject.toml

__init__.py

from nonebot import require
require("nonebot_plugin_session")
require("nonebot_plugin_alconna")
require("nonebot_plugin_localstore")
from nonebot.plugin import PluginMetadata
from .config import cubplugins_permission
from nonebot.plugin import inherit_supported_adapters

__plugin_meta__ = PluginMetadata(
    name="插件响应鉴权",
    description="nonebot简单易用的黑名单插件，实现分插件拉黑用户/群聊/全局",
    usage='''用户操作模式:  #(全局|插件名称)(拉黑|解黑)(全员|用户id)
群聊操作模式: #(全局|插件名称)(封禁群|解禁群)(群号|留空封禁所在群)''',

    type="application",
    # 发布必填，当前有效类型有：`library`（为其他插件编写提供功能），`application`（向机器人用户提供功能）。

    homepage="https://github.com/cubstaryow/nonebot-plugin-authrespond",
    # 发布必填。

    config=cubplugins_permission,
    # 插件配置项类，如无需配置可不填写。
    supported_adapters=inherit_supported_adapters(
        "nonebot_plugin_alconna",
        "nonebot_plugin_session",
        "nonebot_plugin_localstore"
    )
    # 支持的适配器集合，其中 `~` 在此处代表前缀 `nonebot.adapters.`，其余适配器亦按此格式填写。
    # 若插件可以保证兼容所有适配器（即仅使用基本适配器功能）可不填写，否则应该列出插件支持的适配器。
)

from .run import *

from .perm import *

run.py

from nonebot import get_driver , logger
from nonebot.matcher import Matcher
from nonebot.message import run_preprocessor
from nonebot.exception import IgnoredException
from .cubp import cubp
from nonebot_plugin_session import EventSession

superusers = get_driver().config.superusers


@run_preprocessor
async def pass_run(
    matcher: Matcher,
    session: EventSession,
):
    user_id = str(session.id1) # 只能获取？
    if session.level >= 2:
        mode = await passgroup_rule(user_id, session.id2, matcher)
        if mode == "white": return
        if session.level == 3:
             #子ID也支持但是可能会有冲突
            mode = await passgroup_rule(user_id, session.id3, matcher)
            if mode == "white": return
    await pass_rule(user_id, matcher)
           
    pass


# 白名单 - 普通
async def pass_rule(user_id: str, matcher: Matcher):

    modulename = matcher.plugin.name
    if modulename.startswith("nonebot_plugin_"):
        modulename = modulename.replace("nonebot_plugin_", "")
    msg = f"plugin {modulename} is responses by {user_id}, check permission!"
    logger.opt(colors=True).debug(msg)
    check , mode = cubp.checkperm(modulename, user_id)
    if check:
        msg = f"ID {user_id} is not allow run {modulename}"
        logger.opt(colors=True).warning(msg)
        if user_id in superusers:
            return "white"
        raise IgnoredException(msg) from None
    return mode


# 白名单 - 群组
async def passgroup_rule(user_id: str, group_id: str, matcher: Matcher):
    modulename = matcher.plugin.name
    if modulename.startswith("nonebot_plugin_"):
        modulename = modulename.replace("nonebot_plugin_", "")
    msg = f"plugin {modulename} is responses by group {group_id}, check permission!"
    logger.opt(colors=True).debug(msg)
    check , mode = cubp.checkpermgroup(modulename, group_id)
    if check:
        msg = f"ID {group_id} group is not allow run {modulename}"
        logger.opt(colors=True).warning(msg)
        if user_id in superusers:
            return "white"
        raise IgnoredException(msg) from None
    return mode