gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(napcat): 添加事件处理过滤机制，支持黑白名单功能

Browse Source
This commit is contained in:
Windpicker-owo
2025-12-04 21:33:50 +08:00
parent 53bb77686b
commit 43e25378c8
2 changed files with 99 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
src/plugins/built_in/napcat_adapter/plugin.py
View File

@@ -88,6 +88,93 @@ class NapcatAdapter(BaseAdapter):
# 注册 utils 内部使用的适配器实例，便于工具方法自动获取 WS
handler_utils.register_adapter(self)
def _should_process_event(self, raw: Dict[str, Any]) -> bool:
"""
检查事件是否应该被处理（黑白名单过滤）
此方法在 from_platform_message 顶层调用，对所有类型的事件（消息、通知、元事件）进行过滤。
Args:
raw: OneBot 原始事件数据
Returns:
bool: True表示应该处理，False表示应该过滤
"""
if not self.plugin:
return True
plugin_config = self.plugin.config
if not plugin_config:
return True # 如果没有配置，默认处理所有事件
features_config = plugin_config.get("features", {})
post_type = raw.get("post_type")
# 获取用户信息（根据事件类型从不同字段获取）
user_id: str = ""
if post_type == "message":
sender_info = raw.get("sender", {})
user_id = str(sender_info.get("user_id", ""))
elif post_type == "notice":
user_id = str(raw.get("user_id", ""))
else:
# 元事件或其他类型不需要过滤
return True
# 检查全局封禁用户列表
ban_user_ids = [str(item) for item in features_config.get("ban_user_id", [])]
if user_id and user_id in ban_user_ids:
logger.debug(f"用户 {user_id} 在全局封禁列表中，事件被过滤")
return False
# 检查是否屏蔽其他QQ机器人（仅对消息事件生效）
if post_type == "message" and features_config.get("ban_qq_bot", False):
sender_info = raw.get("sender", {})
role = sender_info.get("role", "")
if role == "admin" or "bot" in str(sender_info).lower():
logger.debug(f"检测到机器人消息 {user_id}，事件被过滤")
return False
# 获取消息类型（消息事件使用 message_type，通知事件根据 group_id 判断）
message_type = raw.get("message_type")
group_id = raw.get("group_id")
# 如果是通知事件，根据是否有 group_id 判断是群通知还是私聊通知
if post_type == "notice":
message_type = "group" if group_id else "private"
# 群聊/群通知过滤
if message_type == "group" and group_id:
group_id_str = str(group_id)
group_list_type = features_config.get("group_list_type", "blacklist")
group_list = [str(item) for item in features_config.get("group_list", [])]
if group_list_type == "blacklist":
if group_id_str in group_list:
logger.debug(f"群聊 {group_id_str} 在黑名单中，事件被过滤")
return False
else: # whitelist
if group_id_str not in group_list:
logger.debug(f"群聊 {group_id_str} 不在白名单中，事件被过滤")
return False
# 私聊/私聊通知过滤
elif message_type == "private":
private_list_type = features_config.get("private_list_type", "blacklist")
private_list = [str(item) for item in features_config.get("private_list", [])]
if private_list_type == "blacklist":
if user_id in private_list:
logger.debug(f"私聊用户 {user_id} 在黑名单中，事件被过滤")
return False
else: # whitelist
if user_id not in private_list:
logger.debug(f"私聊用户 {user_id} 不在白名单中，事件被过滤")
return False
# 通过所有过滤条件
return True
async def on_adapter_loaded(self) -> None:
"""适配器加载时的初始化"""
logger.info("Napcat 适配器正在启动...")
@@ -161,6 +248,8 @@ class NapcatAdapter(BaseAdapter):
- notice 事件 → 通知（戳一戳、表情回复等）
- meta_event 事件 → 元事件（心跳、生命周期）
- API 响应 → 存入响应池
注意：黑白名单等过滤机制在此方法最开始执行，确保所有类型的事件都能被过滤。
"""
post_type = raw.get("post_type")
@@ -171,6 +260,11 @@ class NapcatAdapter(BaseAdapter):
future = self._response_pool[echo]
if not future.done():
future.set_result(raw)
return None
# 顶层过滤：黑白名单等过滤机制
if not self._should_process_event(raw):
return None
try:
# 消息事件

84
src/plugins/built_in/napcat_adapter/src/handlers/to_core/message_handler.py
View File

@@ -39,79 +39,6 @@ class MessageHandler:
"""设置插件配置"""
self.plugin_config = config
def _should_process_message(self, raw: Dict[str, Any]) -> bool:
"""
检查消息是否应该被处理（黑白名单过滤）
Args:
raw: OneBot 原始消息数据
Returns:
bool: True表示应该处理，False表示应该过滤
"""
if not self.plugin_config:
return True # 如果没有配置，默认处理所有消息
features_config = self.plugin_config.get("features", {})
# 获取消息基本信息
message_type = raw.get("message_type")
sender_info = raw.get("sender", {})
user_id = str(sender_info.get("user_id", ""))
# 检查全局封禁用户列表
ban_user_ids = [str(item) for item in features_config.get("ban_user_id", [])]
if user_id in ban_user_ids:
logger.debug(f"用户 {user_id} 在全局封禁列表中，消息被过滤")
return False
# 检查是否屏蔽其他QQ机器人
if features_config.get("ban_qq_bot", False):
# 判断是否为机器人消息：通常通过sender中的role字段或其他标识
role = sender_info.get("role", "")
if role == "admin" or "bot" in str(sender_info).lower():
logger.debug(f"检测到机器人消息 {user_id}，消息被过滤")
return False
# 群聊消息处理
if message_type == "group":
group_id = str(raw.get("group_id", ""))
# 获取群聊配置
group_list_type = features_config.get("group_list_type", "blacklist")
group_list = [str(item) for item in features_config.get("group_list", [])]
if group_list_type == "blacklist":
# 黑名单模式：如果在黑名单中就过滤
if group_id in group_list:
logger.debug(f"群聊 {group_id} 在黑名单中，消息被过滤")
return False
else: # whitelist
# 白名单模式：如果不在白名单中就过滤
if group_id not in group_list:
logger.debug(f"群聊 {group_id} 不在白名单中，消息被过滤")
return False
# 私聊消息处理
elif message_type == "private":
# 获取私聊配置
private_list_type = features_config.get("private_list_type", "blacklist")
private_list = [str(item) for item in features_config.get("private_list", [])]
if private_list_type == "blacklist":
# 黑名单模式：如果在黑名单中就过滤
if user_id in private_list:
logger.debug(f"私聊用户 {user_id} 在黑名单中，消息被过滤")
return False
else: # whitelist
# 白名单模式：如果不在白名单中就过滤
if user_id not in private_list:
logger.debug(f"私聊用户 {user_id} 不在白名单中，消息被过滤")
return False
# 通过所有过滤条件
return True
async def handle_raw_message(self, raw: Dict[str, Any]):
"""
处理原始消息并转换为 MessageEnvelope
@@ -120,18 +47,17 @@ class MessageHandler:
raw: OneBot 原始消息数据
Returns:
MessageEnvelope (dict) or None (if message is filtered)
MessageEnvelope (dict) or None
Note:
黑白名单过滤已移动到 NapcatAdapter.from_platform_message 顶层执行，
确保所有类型的事件（消息、通知等）都能被统一过滤。
"""
message_type = raw.get("message_type")
message_id = str(raw.get("message_id", ""))
message_time = time.time()
# 黑白名单过滤
if not self._should_process_message(raw):
logger.debug(f"消息被黑白名单过滤丢弃: message_id={message_id}")
return None
msg_builder = MessageBuilder()
# 构造用户信息