214 lines
8.2 KiB
Python
214 lines
8.2 KiB
Python
"""
|
||
消息类型定义和映射
|
||
|
||
定义个微 API 的消息类型常量，以及到内部事件的映射关系
|
||
"""
|
||
|
||
|
||
class MessageType:
|
||
"""消息类型常量（基于实际测试）"""
|
||
# 系统消息类型
|
||
MT_DEBUG_LOG = 11024 # 调试日志
|
||
MT_USER_LOGIN = 11025 # 用户登录
|
||
MT_USER_LOGOUT = 11026 # 用户登出
|
||
MT_GET_LOGIN_INFO = 11028 # 获取登录信息
|
||
|
||
# 消息通知类型（基于实际测试修正）
|
||
MT_TEXT = 11046 # 文本消息
|
||
MT_IMAGE = 11047 # 图片消息
|
||
MT_VOICE = 11048 # 语音消息
|
||
MT_VIDEO = 11051 # 视频消息
|
||
MT_EMOJI = 11052 # 表情消息
|
||
MT_REVOKE = 11057 # 撤回消息
|
||
MT_SYSTEM = 11058 # 系统消息
|
||
MT_QUOTE = 11061 # 引用消息
|
||
MT_FRIEND_REQUEST = 11056 # 好友请求
|
||
|
||
# 实际测试得出的正确映射
|
||
MT_LOCATION = 11053 # 位置消息（实际）
|
||
MT_LINK = 11054 # 链接消息（实际）
|
||
MT_FILE = 11055 # 文件消息（实际）
|
||
|
||
# 兼容性定义
|
||
MT_CARD = 11055 # 名片消息（临时兼容，实际可能是文件类型）
|
||
MT_MINIAPP = 11054 # 小程序消息（临时兼容，实际可能是链接类型）
|
||
|
||
# 群聊通知类型
|
||
MT_CHATROOM_MEMBER_ADD = 11098 # 群成员新增
|
||
MT_CHATROOM_MEMBER_REMOVE = 11099 # 群成员删除
|
||
MT_CHATROOM_INFO_CHANGE = 11100 # 群信息变化（成员数量变化等）
|
||
|
||
# 发送消息类型
|
||
MT_SEND_TEXT = 11036 # 发送文本
|
||
|
||
|
||
# 消息类型到事件名称的映射
|
||
MESSAGE_TYPE_MAP = {
|
||
MessageType.MT_TEXT: "text_message",
|
||
MessageType.MT_IMAGE: "image_message",
|
||
MessageType.MT_VOICE: "voice_message",
|
||
MessageType.MT_VIDEO: "video_message",
|
||
MessageType.MT_EMOJI: "emoji_message",
|
||
MessageType.MT_REVOKE: "revoke_message",
|
||
MessageType.MT_SYSTEM: "system_message",
|
||
MessageType.MT_FRIEND_REQUEST: "friend_request",
|
||
MessageType.MT_QUOTE: "quote_message",
|
||
MessageType.MT_CHATROOM_MEMBER_ADD: "chatroom_member_add",
|
||
MessageType.MT_CHATROOM_MEMBER_REMOVE: "chatroom_member_remove",
|
||
MessageType.MT_CHATROOM_INFO_CHANGE: "chatroom_info_change",
|
||
|
||
# 修正后的映射（基于实际测试）
|
||
MessageType.MT_LOCATION: "location_message", # 11053 -> 位置消息
|
||
MessageType.MT_LINK: "link_message", # 11054 -> 链接消息
|
||
MessageType.MT_FILE: "file_message", # 11055 -> 文件消息
|
||
}
|
||
|
||
|
||
def normalize_message(msg_type: int, data: dict) -> dict:
|
||
"""
|
||
将个微 API 的消息格式转换为统一的内部格式（兼容 XYBot）
|
||
|
||
Args:
|
||
msg_type: 消息类型
|
||
data: 原始消息数据
|
||
|
||
Returns:
|
||
标准化的消息字典
|
||
"""
|
||
# 基础消息结构
|
||
message = {
|
||
"MsgType": msg_type,
|
||
"FromWxid": data.get("from_wxid", ""),
|
||
"ToWxid": data.get("to_wxid", ""),
|
||
"Content": data.get("msg", data.get("content", data.get("raw_msg", ""))), # 系统消息使用 raw_msg
|
||
"CreateTime": data.get("timestamp", data.get("create_time", 0)),
|
||
"IsGroup": False,
|
||
"SenderWxid": data.get("from_wxid", ""),
|
||
}
|
||
|
||
# 判断是否是群聊消息（room_wxid 不为空）
|
||
room_wxid = data.get("room_wxid", "")
|
||
if room_wxid:
|
||
message["IsGroup"] = True
|
||
message["FromWxid"] = room_wxid
|
||
message["SenderWxid"] = data.get("from_wxid", "")
|
||
|
||
# @ 消息处理
|
||
if "at_user_list" in data:
|
||
message["Ats"] = data["at_user_list"]
|
||
elif "at_list" in data:
|
||
message["Ats"] = data["at_list"]
|
||
|
||
# 图片消息
|
||
if msg_type == MessageType.MT_IMAGE:
|
||
message["ImagePath"] = data.get("image_path", "")
|
||
|
||
# 文件消息（实际类型11055）
|
||
if msg_type == MessageType.MT_FILE:
|
||
message["Filename"] = data.get("filename", "")
|
||
message["FileExtend"] = data.get("file_extend", "")
|
||
message["File"] = data.get("file_data", "")
|
||
|
||
# 语音消息
|
||
if msg_type == MessageType.MT_VOICE:
|
||
message["ImgBuf"] = {"buffer": data.get("voice_data", "")}
|
||
|
||
# 视频消息
|
||
if msg_type == MessageType.MT_VIDEO:
|
||
message["Video"] = data.get("video_data", "")
|
||
|
||
# 引用消息
|
||
if "quote" in data:
|
||
message["Quote"] = data["quote"]
|
||
|
||
# 引用消息的 @ 提取（从 XML 中解析）
|
||
if msg_type == MessageType.MT_QUOTE:
|
||
try:
|
||
import xml.etree.ElementTree as ET
|
||
content = message.get("Content", "")
|
||
if content:
|
||
root = ET.fromstring(content)
|
||
title = root.find(".//title")
|
||
if title is not None and title.text:
|
||
title_text = title.text
|
||
# 检查 title 中是否包含 @
|
||
if "@" in title_text:
|
||
# 从 main_config.toml 读取机器人昵称
|
||
import tomllib
|
||
from pathlib import Path
|
||
config_path = Path("main_config.toml")
|
||
if config_path.exists():
|
||
with open(config_path, "rb") as f:
|
||
main_config = tomllib.load(f)
|
||
bot_nickname = main_config.get("Bot", {}).get("nickname", "")
|
||
bot_wxid = main_config.get("Bot", {}).get("wxid", "")
|
||
|
||
# 检查是否 @ 了机器人
|
||
if bot_nickname and f"@{bot_nickname}" in title_text:
|
||
message["Ats"] = [bot_wxid] if bot_wxid else []
|
||
except Exception:
|
||
pass # 解析失败则忽略
|
||
|
||
# 位置消息（实际类型11053）
|
||
if msg_type == MessageType.MT_LOCATION:
|
||
message["Latitude"] = data.get("latitude", 0)
|
||
message["Longitude"] = data.get("longitude", 0)
|
||
message["LocationTitle"] = data.get("title", "")
|
||
message["LocationAddress"] = data.get("address", "")
|
||
|
||
# 链接消息（实际类型11054）
|
||
if msg_type == MessageType.MT_LINK:
|
||
message["LinkTitle"] = data.get("title", "")
|
||
message["LinkDesc"] = data.get("desc", "")
|
||
message["LinkUrl"] = data.get("url", "")
|
||
message["LinkThumb"] = data.get("thumb_url", "")
|
||
message["MiniappPagePath"] = data.get("page_path", "")
|
||
|
||
# 好友请求
|
||
if msg_type == MessageType.MT_FRIEND_REQUEST:
|
||
message["V3"] = data.get("v3", "")
|
||
message["V4"] = data.get("v4", "")
|
||
message["Scene"] = data.get("scene", 0)
|
||
|
||
# 群成员新增 (type=11098)
|
||
if msg_type == MessageType.MT_CHATROOM_MEMBER_ADD:
|
||
message["FromWxid"] = data.get("room_wxid", "")
|
||
message["IsGroup"] = True
|
||
message["RoomWxid"] = data.get("room_wxid", "")
|
||
message["RoomNickname"] = data.get("nickname", "")
|
||
message["MemberList"] = data.get("member_list", [])
|
||
message["TotalMember"] = data.get("total_member", 0)
|
||
message["ManagerWxid"] = data.get("manager_wxid", "")
|
||
|
||
# 群成员删除 (type=11099)
|
||
if msg_type == MessageType.MT_CHATROOM_MEMBER_REMOVE:
|
||
message["FromWxid"] = data.get("room_wxid", "")
|
||
message["IsGroup"] = True
|
||
message["RoomWxid"] = data.get("room_wxid", "")
|
||
message["RoomNickname"] = data.get("nickname", "")
|
||
message["MemberList"] = data.get("member_list", [])
|
||
message["TotalMember"] = data.get("total_member", 0)
|
||
message["ManagerWxid"] = data.get("manager_wxid", "")
|
||
|
||
# 系统消息 (type=11058)
|
||
if msg_type == MessageType.MT_SYSTEM:
|
||
# 系统消息的内容在 raw_msg 字段
|
||
message["Content"] = data.get("raw_msg", "")
|
||
# 系统消息也可能是群聊消息
|
||
if room_wxid:
|
||
message["IsGroup"] = True
|
||
message["FromWxid"] = room_wxid
|
||
|
||
# 群信息变化 (type=11100)
|
||
if msg_type == MessageType.MT_CHATROOM_INFO_CHANGE:
|
||
message["FromWxid"] = data.get("room_wxid", "")
|
||
message["IsGroup"] = True
|
||
message["RoomWxid"] = data.get("room_wxid", "")
|
||
message["RoomNickname"] = data.get("nickname", "")
|
||
message["TotalMember"] = data.get("total_member", 0)
|
||
message["ManagerWxid"] = data.get("manager_wxid", "")
|
||
# member_list 可能存在也可能不存在
|
||
message["MemberList"] = data.get("member_list", [])
|
||
|
||
return message
|