拦截器
拦截器
主要学习事件拦截。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
链接到当前文件 0
没有文件链接到当前文件