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

feat(napcat): 增加消息与交互相关事件处理模块

Browse Source 
新增多个事件处理器，支持私聊/群聊消息管理、历史记录、戳一戳、合并转发、表情点赞、AI语音等丰富消息功能，统一通过 napcat 接口发送并标准化返回 HandlerResult。

新增事件类型及其文档化注释，涵盖:
- SEND_PRIVATE_MSG / SEND_POKE / DELETE_MSG
- GET_GROUP_MSG_HISTORY / GET_MSG / GET_FORWARD_MSG
- SET_MSG_EMOJI_LIKE / GET_FRIEND_MSG_HISTORY / FETCH_EMOJI_LIKE
- SEND_FORWARD_MSG / SEND_GROUP_AI_RECORD

移除冗余空行并修正枚举名称拼写（FORWARF_MSG → FORWARD_MSG），为使用者提供完整入参和返回值说明。
This commit is contained in:
Windpicker-owo
2025-08-30 02:15:12 +08:00
parent cb93b38162
commit c6a0522316
2 changed files with 741 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

388
plugins/napcat_adapter_plugin/event_handlers.py
View File

@@ -1,4 +1,3 @@
from src.plugin_system import BaseEventHandler
from src.plugin_system.base.base_event import HandlerResult
@@ -548,3 +547,390 @@ class SetDiyOnlineStatusHandler(BaseEventHandler):
else:
logger.error("事件 napcat_set_diy_online_status 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class SendPrivateMsgHandler(BaseEventHandler):
handler_name: str = "napcat_send_private_msg_handler"
handler_description: str = "发送私聊消息"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.SEND_PRIVATE_MSG]
async def execute(self, params: dict):
raw = params.get("raw", {})
user_id = params.get("user_id", "")
message = params.get("message", "")
if params.get("raw", ""):
user_id = raw.get("user_id", "")
message = raw.get("message", "")
if not user_id or not message:
logger.error("事件 napcat_send_private_msg 缺少必要参数: user_id 或 message")
return HandlerResult(False, False, {"status": "error"})
payload = {
"user_id": str(user_id),
"message": message
}
response = await send_handler.send_message_to_napcat(action="send_private_msg", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_send_private_msg 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class SendPokeHandler(BaseEventHandler):
handler_name: str = "napcat_send_poke_handler"
handler_description: str = "发送戳一戳"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.SEND_POKE]
async def execute(self, params: dict):
raw = params.get("raw", {})
user_id = params.get("user_id", "")
group_id = params.get("group_id", None)
if params.get("raw", ""):
user_id = raw.get("user_id", "")
group_id = raw.get("group_id", None)
if not user_id:
logger.error("事件 napcat_send_poke 缺少必要参数: user_id")
return HandlerResult(False, False, {"status": "error"})
payload = {
"user_id": str(user_id)
}
if group_id is not None:
payload["group_id"] = str(group_id)
response = await send_handler.send_message_to_napcat(action="send_poke", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_send_poke 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class DeleteMsgHandler(BaseEventHandler):
handler_name: str = "napcat_delete_msg_handler"
handler_description: str = "撤回消息"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.DELETE_MSG]
async def execute(self, params: dict):
raw = params.get("raw", {})
message_id = params.get("message_id", "")
if params.get("raw", ""):
message_id = raw.get("message_id", "")
if not message_id:
logger.error("事件 napcat_delete_msg 缺少必要参数: message_id")
return HandlerResult(False, False, {"status": "error"})
payload = {
"message_id": str(message_id)
}
response = await send_handler.send_message_to_napcat(action="delete_msg", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_delete_msg 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class GetGroupMsgHistoryHandler(BaseEventHandler):
handler_name: str = "napcat_get_group_msg_history_handler"
handler_description: str = "获取群历史消息"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.GET_GROUP_MSG_HISTORY]
async def execute(self, params: dict):
raw = params.get("raw", {})
group_id = params.get("group_id", "")
message_seq = params.get("message_seq", 0)
count = params.get("count", 20)
reverseOrder = params.get("reverseOrder", False)
if params.get("raw", ""):
group_id = raw.get("group_id", "")
message_seq = raw.get("message_seq", 0)
count = raw.get("count", 20)
reverseOrder = raw.get("reverseOrder", False)
if not group_id:
logger.error("事件 napcat_get_group_msg_history 缺少必要参数: group_id")
return HandlerResult(False, False, {"status": "error"})
payload = {
"group_id": str(group_id),
"message_seq": int(message_seq),
"count": int(count),
"reverseOrder": bool(reverseOrder)
}
response = await send_handler.send_message_to_napcat(action="get_group_msg_history", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_get_group_msg_history 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class GetMsgHandler(BaseEventHandler):
handler_name: str = "napcat_get_msg_handler"
handler_description: str = "获取消息详情"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.GET_MSG]
async def execute(self, params: dict):
raw = params.get("raw", {})
message_id = params.get("message_id", "")
if params.get("raw", ""):
message_id = raw.get("message_id", "")
if not message_id:
logger.error("事件 napcat_get_msg 缺少必要参数: message_id")
return HandlerResult(False, False, {"status": "error"})
payload = {
"message_id": str(message_id)
}
response = await send_handler.send_message_to_napcat(action="get_msg", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_get_msg 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class GetForwardMsgHandler(BaseEventHandler):
handler_name: str = "napcat_get_forward_msg_handler"
handler_description: str = "获取合并转发消息"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.GET_FORWARD_MSG]
async def execute(self, params: dict):
raw = params.get("raw", {})
message_id = params.get("message_id", "")
if params.get("raw", ""):
message_id = raw.get("message_id", "")
if not message_id:
logger.error("事件 napcat_get_forward_msg 缺少必要参数: message_id")
return HandlerResult(False, False, {"status": "error"})
payload = {
"message_id": str(message_id)
}
response = await send_handler.send_message_to_napcat(action="get_forward_msg", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_get_forward_msg 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class SetMsgEmojiLikeHandler(BaseEventHandler):
handler_name: str = "napcat_set_msg_emoji_like_handler"
handler_description: str = "贴表情"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.SET_MSG_EMOJI_LIKE]
async def execute(self, params: dict):
raw = params.get("raw", {})
message_id = params.get("message_id", "")
emoji_id = params.get("emoji_id", 0)
set_flag = params.get("set", True)
if params.get("raw", ""):
message_id = raw.get("message_id", "")
emoji_id = raw.get("emoji_id", 0)
set_flag = raw.get("set", True)
if not message_id or emoji_id is None or set_flag is None:
logger.error("事件 napcat_set_msg_emoji_like 缺少必要参数")
return HandlerResult(False, False, {"status": "error"})
payload = {
"message_id": str(message_id),
"emoji_id": int(emoji_id),
"set": bool(set_flag)
}
response = await send_handler.send_message_to_napcat(action="set_msg_emoji_like", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_set_msg_emoji_like 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class GetFriendMsgHistoryHandler(BaseEventHandler):
handler_name: str = "napcat_get_friend_msg_history_handler"
handler_description: str = "获取好友历史消息"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.GET_FRIEND_MSG_HISTORY]
async def execute(self, params: dict):
raw = params.get("raw", {})
user_id = params.get("user_id", "")
message_seq = params.get("message_seq", 0)
count = params.get("count", 20)
reverseOrder = params.get("reverseOrder", False)
if params.get("raw", ""):
user_id = raw.get("user_id", "")
message_seq = raw.get("message_seq", 0)
count = raw.get("count", 20)
reverseOrder = raw.get("reverseOrder", False)
if not user_id:
logger.error("事件 napcat_get_friend_msg_history 缺少必要参数: user_id")
return HandlerResult(False, False, {"status": "error"})
payload = {
"user_id": str(user_id),
"message_seq": int(message_seq),
"count": int(count),
"reverseOrder": bool(reverseOrder)
}
response = await send_handler.send_message_to_napcat(action="get_friend_msg_history", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_get_friend_msg_history 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class FetchEmojiLikeHandler(BaseEventHandler):
handler_name: str = "napcat_fetch_emoji_like_handler"
handler_description: str = "获取贴表情详情"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.FETCH_EMOJI_LIKE]
async def execute(self, params: dict):
raw = params.get("raw", {})
message_id = params.get("message_id", "")
emoji_id = params.get("emoji_id", "")
emoji_type = params.get("emoji_type", "")
count = params.get("count", 20)
if params.get("raw", ""):
message_id = raw.get("message_id", "")
emoji_id = raw.get("emoji_id", "")
emoji_type = raw.get("emoji_type", "")
count = raw.get("count", 20)
if not message_id or not emoji_id or not emoji_type:
logger.error("事件 napcat_fetch_emoji_like 缺少必要参数")
return HandlerResult(False, False, {"status": "error"})
payload = {
"message_id": str(message_id),
"emojiId": str(emoji_id),
"emojiType": str(emoji_type),
"count": int(count)
}
response = await send_handler.send_message_to_napcat(action="fetch_emoji_like", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_fetch_emoji_like 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class SendForwardMsgHandler(BaseEventHandler):
handler_name: str = "napcat_send_forward_msg_handler"
handler_description: str = "发送合并转发消息"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.SEND_FORWARF_MSG]
async def execute(self, params: dict):
raw = params.get("raw", {})
messages = params.get("messages", {})
news = params.get("news", {})
prompt = params.get("prompt", "")
summary = params.get("summary", "")
source = params.get("source", "")
group_id = params.get("group_id", None)
user_id = params.get("user_id", None)
if params.get("raw", ""):
messages = raw.get("messages", {})
news = raw.get("news", {})
prompt = raw.get("prompt", "")
summary = raw.get("summary", "")
source = raw.get("source", "")
group_id = raw.get("group_id", None)
user_id = raw.get("user_id", None)
if not messages or not news or not prompt or not summary or not source:
logger.error("事件 napcat_send_forward_msg 缺少必要参数")
return HandlerResult(False, False, {"status": "error"})
payload = {
"messages": messages,
"news": news,
"prompt": prompt,
"summary": summary,
"source": source
}
if group_id is not None:
payload["group_id"] = str(group_id)
if user_id is not None:
payload["user_id"] = str(user_id)
response = await send_handler.send_message_to_napcat(action="send_forward_msg", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_send_forward_msg 请求失败！")
return HandlerResult(False, False, {"status": "error"})
class SendGroupAiRecordHandler(BaseEventHandler):
handler_name: str = "napcat_send_group_ai_record_handler"
handler_description: str = "发送群AI语音"
weight: int = 100
intercept_message: bool = False
init_subscribe = [NapcatEvent.MESSAGE.SEND_GROUP_AI_RECORD]
async def execute(self, params: dict):
raw = params.get("raw", {})
group_id = params.get("group_id", "")
character = params.get("character", "")
text = params.get("text", "")
if params.get("raw", ""):
group_id = raw.get("group_id", "")
character = raw.get("character", "")
text = raw.get("text", "")
if not group_id or not character or not text:
logger.error("事件 napcat_send_group_ai_record 缺少必要参数")
return HandlerResult(False, False, {"status": "error"})
payload = {
"group_id": str(group_id),
"character": character,
"text": text
}
response = await send_handler.send_message_to_napcat(action="send_group_ai_record", params=payload)
if response.get("status", "") == "ok":
return HandlerResult(True, True, response)
else:
logger.error("事件 napcat_send_group_ai_record 请求失败！")
return HandlerResult(False, False, {"status": "error"})

370
plugins/napcat_adapter_plugin/event_types.py
View File

@@ -564,32 +564,370 @@ class NapcatEvent(Enum):
"""
该分类是对信息相关的操作，只能由外部触发，napcat_plugin负责处理
"""
SEND_GROUP_POKE = "napcat_send_group_poke"
'''发送群聊戳一戳'''
SEND_PRIVATE_MSG = "napcat_send_private_msg"
'''发送私聊消息'''
SEND_POKE = "napcat_send_friend_poke"
'''发送戳一戳'''
'''发送私聊消息
Args:
user_id (Opetional[str|int]): 用户id(必需)
message (Opetional[str]): 消息object(必需)
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": {
"message_id": 0
},
"message": "string",
"wording": "string",
"echo": "string"
}
'''
SEND_POKE = "napcat_send_poke"
'''发送戳一戳
Args:
group_id (Optional[str|int]): 群号
user_id (Optional[str|int]): 对方QQ号(必需)
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": null,
"message": "string",
"wording": "string",
"echo": "string"
}
'''
DELETE_MSG = "napcat_delete_msg"
'''撤回消息'''
'''撤回消息
Args:
message_id (Optional[str|int]): 消息id(必需)
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": null,
"message": "string",
"wording": "string",
"echo": "string"
}
'''
GET_GROUP_MSG_HISTORY = "napcat_get_group_msg_history"
'''获取群历史消息'''
'''获取群历史消息
Args:
group_id (Optional[str|int]): 群号(必需)
message_seq (Optional[str|int]): 消息序号,0为最新
count (Optional[int]): 获取数量
reverseOrder (Optional[bool]): 是否倒序
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": {
"messages": [
{
"self_id": 0,
"user_id": 0,
"time": 0,
"message_id": 0,
"message_seq": 0,
"real_id": 0,
"real_seq": "string",
"message_type": "string",
"sender": {
"user_id": 0,
"nickname": "string",
"sex": "male",
"age": 0,
"card": "string",
"role": "owner"
},
"raw_message": "string",
"font": 0,
"sub_type": "string",
"message": [
{
"type": "text",
"data": {
"text": "string"
}
}
],
"message_format": "string",
"post_type": "string",
"group_id": 0
}
]
},
"message": "string",
"wording": "string",
"echo": "string"
}
'''
GET_MSG = "napcat_get_msg"
'''获取消息详情'''
'''获取消息详情
Args:
message_id (Optional[str|int]): 消息id(必需)
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": {
"self_id": 0,
"user_id": 0,
"time": 0,
"message_id": 0,
"message_seq": 0,
"real_id": 0,
"real_seq": "string",
"message_type": "string",
"sender": {
"user_id": 0,
"nickname": "string",
"sex": "male",
"age": 0,
"card": "string",
"role": "owner"
},
"raw_message": "string",
"font": 0,
"sub_type": "string",
"message": [
{
"type": "text",
"data": {
"text": "string"
}
}
],
"message_format": "string",
"post_type": "string",
"group_id": 0
},
"message": "string",
"wording": "string",
"echo": "string"
}
'''
GET_FORWARD_MSG = "napcat_get_forward_msg"
'''获取合并转发消息'''
'''获取合并转发消息
Args:
message_id (Optional[str|int]): 消息id(必需)
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": {
"messages": [
{
"self_id": 0,
"user_id": 0,
"time": 0,
"message_id": 0,
"message_seq": 0,
"real_id": 0,
"real_seq": "string",
"message_type": "string",
"sender": {
"user_id": 0,
"nickname": "string",
"sex": "male",
"age": 0,
"card": "string",
"role": "owner"
},
"raw_message": "string",
"font": 0,
"sub_type": "string",
"message": [
{
"type": "text",
"data": {
"text": "string"
}
}
],
"message_format": "string",
"post_type": "string",
"group_id": 0
}
]
},
"message": "string",
"wording": "string",
"echo": "string"
}
'''
SET_MSG_EMOJI_LIKE = "napcat_set_msg_emoji_like"
'''贴表情'''
'''贴表情
Args:
message_id (Optional[str|int]): 消息id(必需)
emoji_id (Optional[int]): 表情id(必需)
set (Optional[bool]): 是否贴(必需)
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": {
"result": 0,
"errMsg": "string"
},
"message": "string",
"wording": "string",
"echo": "string"
}
'''
GET_FRIEND_MSG_HISTORY = "napcat_get_friend_msg_history"
'''获取好友历史消息'''
'''获取好友历史消息
Args:
user_id (Optional[str|int]): 用户id(必需)
message_seq (Optional[str|int]): 消息序号,0为最新
count (Optional[int]): 获取数量
reverseOrder (Optional[bool]): 是否倒序
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": {
"messages": [
{
"self_id": 0,
"user_id": 0,
"time": 0,
"message_id": 0,
"message_seq": 0,
"real_id": 0,
"real_seq": "string",
"message_type": "string",
"sender": {
"user_id": 0,
"nickname": "string",
"sex": "male",
"age": 0,
"card": "string",
"role": "owner"
},
"raw_message": "string",
"font": 0,
"sub_type": "string",
"message": [
{
"type": "text",
"data": {
"text": "string"
}
}
],
"message_format": "string",
"post_type": "string",
"group_id": 0
}
]
},
"message": "string",
"wording": "string",
"echo": "string"
}
'''
FETCH_EMOJI_LIKE = "napcat_fetch_emoji_like"
'''获取贴表情详情'''
'''获取贴表情详情
Args:
message_id (Optional[str|int]): 消息id(必需)
emojiId (Optional[str]): 表情id(必需)
emojiType (Optional[str]): 表情类型(必需)
count (Optional[int]): 返回数量
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": {
"result": 0,
"errMsg": "string",
"emojiLikesList": [
{
"tinyId": "string",
"nickName": "string",
"headUrl": "string"
}
],
"cookie": "string",
"isLastPage": true,
"isFirstPage": true
},
"message": "string",
"wording": "string",
"echo": "string"
}
'''
SEND_FORWARF_MSG = "napcat_send_forward_msg"
'''发送合并转发消息'''
GET_RECOED = "napcat_get_record"
'''获取语音消息详情'''
'''发送合并转发消息
Args:
group_id (Optional[str|int]): 群号
user_id (Optional[str|int]): 用户id
messages (Optional[dict]): 一级合并转发消息节点(必需)
news (Optional[dict]): 原转发消息之外的消息(必需)
prompt (Optional[str]): 外显(必需)
summary (Optional[str]): 底下文本(必需)
source (Optional[str]): 内容(必需)
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": {},
"message": "string",
"wording": "string",
"echo": "string"
}
'''
SEND_GROUP_AI_RECORD = "napcat_send_group_ai_record"
'''发送群AI语音'''
'''发送群AI语音
Args:
group_id (Optional[str|int]): 群号(必需)
character (Optional[str]): 角色id(必需)
text (Optional[str]): 文本(必需)
raw (Optional[dict]): 原始请求体
Returns:
dict: {
"status": "ok",
"retcode": 0,
"data": {
"message_id": "string"
},
"message": "string",
"wording": "string",
"echo": "string"
}
'''
class GROUP(Enum):
"""