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

feat: 添加通知处理功能，支持戳一戳、表情回复、群禁言等事件

Browse Source
This commit is contained in:
Windpicker-owo
2025-11-27 22:08:08 +08:00
parent ee39ba0fd2
commit 26520c123a
2 changed files with 543 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
src/plugins/built_in/napcat_adapter/src/event_types.py Normal file
View File

@@ -0,0 +1,37 @@
"""Napcat 适配器事件类型定义"""
class NapcatEvent:
"""Napcat 适配器事件类型"""
class ON_RECEIVED:
"""接收事件"""
FRIEND_INPUT = "napcat.on_received.friend_input" # 好友正在输入
EMOJI_LIEK = "napcat.on_received.emoji_like" # 表情回复（注意：保持原来的拼写）
POKE = "napcat.on_received.poke" # 戳一戳
GROUP_UPLOAD = "napcat.on_received.group_upload" # 群文件上传
GROUP_BAN = "napcat.on_received.group_ban" # 群禁言
GROUP_LIFT_BAN = "napcat.on_received.group_lift_ban" # 群解禁
FRIEND_RECALL = "napcat.on_received.friend_recall" # 好友消息撤回
GROUP_RECALL = "napcat.on_received.group_recall" # 群消息撤回
class MESSAGE:
"""消息相关事件"""
GET_MSG = "napcat.message.get_msg" # 获取消息
class GROUP:
"""群组相关事件"""
SET_GROUP_BAN = "napcat.group.set_group_ban" # 设置群禁言
SET_GROUP_WHOLE_BAN = "napcat.group.set_group_whole_ban" # 设置全员禁言
SET_GROUP_KICK = "napcat.group.set_group_kick" # 踢出群聊
class FRIEND:
"""好友相关事件"""
SEND_LIKE = "napcat.friend.send_like" # 发送点赞
__all__ = ["NapcatEvent"]

523
src/plugins/built_in/napcat_adapter/src/handlers/to_core/notice_handler.py
View File

@@ -2,40 +2,529 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Dict, Optional
import time
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
from mofox_wire import MessageBuilder, SegPayload, UserInfoPayload
from src.common.logger import get_logger
from src.plugin_system.apis import config_api
from ...event_models import ACCEPT_FORMAT, NoticeType, QQ_FACE, PLUGIN_NAME
from ..utils import get_group_info, get_member_info, get_self_info, get_stranger_info, get_message_detail
if TYPE_CHECKING:
from ...plugin import NapcatAdapter
from ....plugin import NapcatAdapter
logger = get_logger("napcat_adapter")
class NoticeHandler:
"""处理 Napcat 通知事件（戳一戳、表情回复等）"""
"""处理 Napcat 通知事件（戳一戳、表情回复、禁言、文件上传等）"""
def __init__(self, adapter: "NapcatAdapter"):
self.adapter = adapter
self.plugin_config: Optional[Dict[str, Any]] = None
# 戳一戳防抖时间戳
self.last_poke_time: float = 0.0
def set_plugin_config(self, config: Dict[str, Any]) -> None:
"""设置插件配置"""
self.plugin_config = config
def _get_config(self, key: str, default: Any = None) -> Any:
"""获取插件配置的辅助方法"""
if not self.plugin_config:
return default
return config_api.get_plugin_config(self.plugin_config, key, default)
async def handle_notice(self, raw: Dict[str, Any]):
"""处理通知事件"""
# 简化版本：返回一个空的 MessageEnvelope
import time
import uuid
return {
"direction": "incoming",
"message_info": {
"platform": "qq",
"message_id": str(uuid.uuid4()),
"time": time.time(),
},
"message_segment": {"type": "text", "data": "[通知事件]"},
"timestamp_ms": int(time.time() * 1000),
"""
处理通知事件
Args:
raw: OneBot 原始通知数据
Returns:
MessageEnvelope (dict) or None
"""
notice_type = raw.get("notice_type")
message_time: float = time.time()
self_id = raw.get("self_id")
group_id = raw.get("group_id")
user_id = raw.get("user_id")
target_id = raw.get("target_id")
handled_segment: SegPayload | None = None
user_info: UserInfoPayload | None = None
system_notice: bool = False
notice_config: Dict[str, Any] = {
"is_notice": False,
"is_public_notice": False,
"target_id": target_id,
}
match notice_type:
case NoticeType.friend_recall:
logger.info("好友撤回一条消息")
logger.info(f"撤回消息ID：{raw.get('message_id')}, 撤回时间：{raw.get('time')}")
logger.warning("暂时不支持撤回消息处理")
return None
case NoticeType.group_recall:
logger.info("群内用户撤回一条消息")
logger.info(f"撤回消息ID：{raw.get('message_id')}, 撤回时间：{raw.get('time')}")
logger.warning("暂时不支持撤回消息处理")
return None
case NoticeType.notify:
sub_type = raw.get("sub_type")
match sub_type:
case NoticeType.Notify.poke:
if self._get_config("features.enable_poke", True):
logger.debug("处理戳一戳消息")
handled_segment, user_info = await self._handle_poke_notify(raw, group_id, user_id)
if handled_segment and user_info:
notice_config["notice_type"] = "poke"
notice_config["is_notice"] = True
else:
logger.warning("戳一戳消息被禁用，取消戳一戳处理")
return None
case NoticeType.Notify.input_status:
from src.plugin_system.core.event_manager import event_manager
from ...event_types import NapcatEvent
await event_manager.trigger_event(
NapcatEvent.ON_RECEIVED.FRIEND_INPUT,
permission_group=PLUGIN_NAME
)
return None
case _:
logger.warning(f"不支持的notify类型: {notice_type}.{sub_type}")
return None
case NoticeType.group_msg_emoji_like:
if self._get_config("features.enable_emoji_like", True):
logger.debug("处理群聊表情回复")
handled_segment, user_info = await self._handle_group_emoji_like_notify(
raw, group_id, user_id
)
if handled_segment and user_info:
notice_config["notice_type"] = "emoji_like"
notice_config["is_notice"] = True
else:
logger.warning("群聊表情回复被禁用，取消群聊表情回复处理")
return None
case NoticeType.group_ban:
sub_type = raw.get("sub_type")
match sub_type:
case NoticeType.GroupBan.ban:
logger.info("处理群禁言")
handled_segment, user_info = await self._handle_ban_notify(raw, group_id)
if handled_segment and user_info:
system_notice = True
user_id_in_ban = raw.get("user_id")
if user_id_in_ban == 0:
notice_config["notice_type"] = "group_whole_ban"
else:
notice_config["notice_type"] = "group_ban"
notice_config["is_notice"] = True
case NoticeType.GroupBan.lift_ban:
logger.info("处理解除群禁言")
handled_segment, user_info = await self._handle_lift_ban_notify(raw, group_id)
if handled_segment and user_info:
system_notice = True
user_id_in_ban = raw.get("user_id")
if user_id_in_ban == 0:
notice_config["notice_type"] = "group_whole_lift_ban"
else:
notice_config["notice_type"] = "group_lift_ban"
notice_config["is_notice"] = True
case _:
logger.warning(f"不支持的group_ban类型: {notice_type}.{sub_type}")
return None
case NoticeType.group_upload:
logger.info("群文件上传")
if user_id == self_id:
logger.info("检测到机器人自己上传文件，忽略此通知")
return None
handled_segment, user_info = await self._handle_group_upload_notify(
raw, group_id, user_id, self_id
)
if handled_segment and user_info:
notice_config["notice_type"] = "group_upload"
notice_config["is_notice"] = True
case _:
logger.warning(f"不支持的notice类型: {notice_type}")
return None
if not handled_segment or not user_info:
logger.warning("notice处理失败或不支持")
return None
# 使用 MessageBuilder 构建消息
msg_builder = MessageBuilder()
(
msg_builder.direction("incoming")
.message_id("notice")
.timestamp_ms(int(message_time * 1000))
.from_user(
user_id=str(user_info.get("user_id", "")),
platform="qq",
nickname=user_info.get("user_nickname", ""),
cardname=user_info.get("user_cardname", ""),
)
)
# 如果是群消息，添加群信息
if group_id:
fetched_group_info = await get_group_info(group_id)
group_name: str | None = None
if fetched_group_info:
group_name = fetched_group_info.get("group_name")
else:
logger.warning("无法获取notice消息所在群的名称")
msg_builder.from_group(
group_id=str(group_id),
platform="qq",
name=group_name or "",
)
# 设置格式信息
content_format = [handled_segment.get("type", "text")]
if "notify" not in content_format:
content_format.append("notify")
msg_builder.format_info(
content_format=content_format,
accept_format=ACCEPT_FORMAT,
)
# 设置消息段
msg_builder.seg_list([handled_segment])
# 设置 additional_config（包含 notice 相关配置）
res = msg_builder.build()["message_info"]["additional_config"] = notice_config
return res
async def _handle_poke_notify(
self, raw: Dict[str, Any], group_id: Any, user_id: Any
) -> Tuple[SegPayload | None, UserInfoPayload | None]:
"""处理戳一戳通知"""
self_info: dict | None = await get_self_info()
if not self_info:
logger.error("自身信息获取失败")
return None, None
self_id = raw.get("self_id")
target_id = raw.get("target_id")
# 防抖检查：如果是针对机器人的戳一戳，检查防抖时间
if self_id == target_id:
current_time = time.time()
debounce_seconds = self._get_config("features.poke_debounce_seconds", 2.0)
if self.last_poke_time > 0:
time_diff = current_time - self.last_poke_time
if time_diff < debounce_seconds:
logger.debug(
f"戳一戳防抖：用户 {user_id} 的戳一戳被忽略（距离上次戳一戳 {time_diff:.2f} 秒）"
)
return None, None
self.last_poke_time = current_time
target_name: str | None = None
raw_info: list = raw.get("raw_info", [])
if group_id:
user_qq_info: dict | None = await get_member_info(group_id, user_id)
else:
user_qq_info: dict | None = await get_stranger_info(user_id)
if user_qq_info:
user_name = user_qq_info.get("nickname", "QQ用户")
user_cardname = user_qq_info.get("card", "")
else:
user_name = "QQ用户"
user_cardname = ""
logger.debug("无法获取戳一戳对方的用户昵称")
# 计算显示名称
display_name = ""
if self_id == target_id:
target_name = self_info.get("nickname", "")
elif self_id == user_id:
# 不发送机器人戳别人的消息
return None, None
else:
# 如果配置为忽略不是针对自己的戳一戳，则直接返回None
if self._get_config("features.ignore_non_self_poke", False):
logger.debug("忽略不是针对自己的戳一戳消息")
return None, None
if group_id:
fetched_member_info: dict | None = await get_member_info(group_id, target_id)
if fetched_member_info:
target_name = fetched_member_info.get("nickname", "QQ用户")
else:
target_name = "QQ用户"
logger.debug("无法获取被戳一戳方的用户昵称")
display_name = user_name
else:
return None, None
# 解析戳一戳文本
first_txt: str = "戳了戳"
second_txt: str = ""
try:
if len(raw_info) > 2:
first_txt = raw_info[2].get("txt", "戳了戳")
if len(raw_info) > 4:
second_txt = raw_info[4].get("txt", "")
except Exception as e:
logger.warning(f"解析戳一戳消息失败: {str(e)}，将使用默认文本")
user_info: UserInfoPayload = {
"platform": "qq",
"user_id": str(user_id),
"user_nickname": user_name,
"user_cardname": user_cardname,
}
seg_data: SegPayload = {
"type": "text",
"data": f"{display_name}{first_txt}{target_name}{second_txt}（这是QQ的一个功能，用于提及某人，但没那么明显）",
}
return seg_data, user_info
async def _handle_group_emoji_like_notify(
self, raw: Dict[str, Any], group_id: Any, user_id: Any
) -> Tuple[SegPayload | None, UserInfoPayload | None]:
"""处理群聊表情回复通知"""
if not group_id:
logger.error("群ID不能为空，无法处理群聊表情回复通知")
return None, None
user_qq_info: dict | None = await get_member_info(group_id, user_id)
if user_qq_info:
user_name = user_qq_info.get("nickname", "QQ用户")
user_cardname = user_qq_info.get("card", "")
else:
user_name = "QQ用户"
user_cardname = ""
logger.debug("无法获取表情回复对方的用户昵称")
# 触发事件
from src.plugin_system.core.event_manager import event_manager
from ...event_types import NapcatEvent
target_message = await get_message_detail(raw.get("message_id", ""))
target_message_text = ""
if target_message:
target_message_text = target_message.get("raw_message", "")
else:
logger.error("未找到对应消息")
return None, None
if len(target_message_text) > 15:
target_message_text = target_message_text[:15] + "..."
user_info: UserInfoPayload = {
"platform": "qq",
"user_id": str(user_id),
"user_nickname": user_name,
"user_cardname": user_cardname,
}
likes_list = raw.get("likes", [])
like_emoji_id = ""
if likes_list and len(likes_list) > 0:
like_emoji_id = str(likes_list[0].get("emoji_id", ""))
# 触发表情回复事件
await event_manager.trigger_event(
NapcatEvent.ON_RECEIVED.EMOJI_LIEK,
permission_group=PLUGIN_NAME,
group_id=group_id,
user_id=user_id,
message_id=raw.get("message_id", ""),
emoji_id=like_emoji_id,
)
emoji_text = QQ_FACE.get(like_emoji_id, f"[表情{like_emoji_id}]")
seg_data: SegPayload = {
"type": "text",
"data": f"{user_name}使用Emoji表情{emoji_text}回应了消息[{target_message_text}]",
}
return seg_data, user_info
async def _handle_group_upload_notify(
self, raw: Dict[str, Any], group_id: Any, user_id: Any, self_id: Any
) -> Tuple[SegPayload | None, UserInfoPayload | None]:
"""处理群文件上传通知"""
if not group_id:
logger.error("群ID不能为空，无法处理群文件上传通知")
return None, None
user_qq_info: dict | None = await get_member_info(group_id, user_id)
if user_qq_info:
user_name = user_qq_info.get("nickname", "QQ用户")
user_cardname = user_qq_info.get("card", "")
else:
user_name = "QQ用户"
user_cardname = ""
logger.debug("无法获取上传文件的用户昵称")
file_info = raw.get("file")
if not file_info:
logger.error("群文件上传通知中缺少文件信息")
return None, None
user_info: UserInfoPayload = {
"platform": "qq",
"user_id": str(user_id),
"user_nickname": user_name,
"user_cardname": user_cardname,
}
file_name = file_info.get("name", "未知文件")
file_size = file_info.get("size", 0)
seg_data: SegPayload = {
"type": "text",
"data": f"{user_name} 上传了文件: {file_name} (大小: {file_size} 字节)",
}
return seg_data, user_info
async def _handle_ban_notify(
self, raw: Dict[str, Any], group_id: Any
) -> Tuple[SegPayload | None, UserInfoPayload | None]:
"""处理群禁言通知"""
if not group_id:
logger.error("群ID不能为空，无法处理禁言通知")
return None, None
# 获取操作者信息
operator_id = raw.get("operator_id")
operator_nickname: str = "QQ用户"
operator_cardname: str = ""
member_info: dict | None = await get_member_info(group_id, operator_id)
if member_info:
operator_nickname = member_info.get("nickname", "QQ用户")
operator_cardname = member_info.get("card", "")
else:
logger.warning("无法获取禁言执行者的昵称，消息可能会无效")
operator_info: UserInfoPayload = {
"platform": "qq",
"user_id": str(operator_id),
"user_nickname": operator_nickname,
"user_cardname": operator_cardname,
}
# 获取被禁言者信息
user_id = raw.get("user_id")
banned_user_info: Dict[str, Any] | None = None
user_nickname: str = "QQ用户"
user_cardname: str = ""
sub_type: str = ""
duration = raw.get("duration")
if duration is None:
logger.error("禁言时长不能为空，无法处理禁言通知")
return None, None
if user_id == 0: # 全体禁言
sub_type = "whole_ban"
else: # 单人禁言
sub_type = "ban"
fetched_member_info: dict | None = await get_member_info(group_id, user_id)
if fetched_member_info:
user_nickname = fetched_member_info.get("nickname", "QQ用户")
user_cardname = fetched_member_info.get("card", "")
banned_user_info = {
"platform": "qq",
"user_id": str(user_id),
"user_nickname": user_nickname,
"user_cardname": user_cardname,
}
seg_data: SegPayload = {
"type": "notify",
"data": {
"sub_type": sub_type,
"duration": duration,
"banned_user_info": banned_user_info,
},
}
return seg_data, operator_info
async def _handle_lift_ban_notify(
self, raw: Dict[str, Any], group_id: Any
) -> Tuple[SegPayload | None, UserInfoPayload | None]:
"""处理解除群禁言通知"""
if not group_id:
logger.error("群ID不能为空，无法处理解除禁言通知")
return None, None
# 获取操作者信息
operator_id = raw.get("operator_id")
operator_nickname: str = "QQ用户"
operator_cardname: str = ""
member_info: dict | None = await get_member_info(group_id, operator_id)
if member_info:
operator_nickname = member_info.get("nickname", "QQ用户")
operator_cardname = member_info.get("card", "")
else:
logger.warning("无法获取解除禁言执行者的昵称，消息可能会无效")
operator_info: UserInfoPayload = {
"platform": "qq",
"user_id": str(operator_id),
"user_nickname": operator_nickname,
"user_cardname": operator_cardname,
}
# 获取被解除禁言者信息
sub_type: str = ""
user_nickname: str = "QQ用户"
user_cardname: str = ""
lifted_user_info: Dict[str, Any] | None = None
user_id = raw.get("user_id")
if user_id == 0: # 全体禁言解除
sub_type = "whole_lift_ban"
else: # 单人禁言解除
sub_type = "lift_ban"
fetched_member_info: dict | None = await get_member_info(group_id, user_id)
if fetched_member_info:
user_nickname = fetched_member_info.get("nickname", "QQ用户")
user_cardname = fetched_member_info.get("card", "")
else:
logger.warning("无法获取解除禁言消息发送者的昵称，消息可能会无效")
lifted_user_info = {
"platform": "qq",
"user_id": str(user_id),
"user_nickname": user_nickname,
"user_cardname": user_cardname,
}
seg_data: SegPayload = {
"type": "notify",
"data": {
"sub_type": sub_type,
"lifted_user_info": lifted_user_info,
}
}
return seg_data, operator_info