""" 消息类型定义和映射 定义个微 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