""" 消息类型定义和映射 定义新协议 API 的消息类型常量，以及到内部事件的映射关系 """ import json import xml.etree.ElementTree as ET 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 = 11062 # 名片消息 MT_MINIAPP = 11063 # 小程序消息 MT_UNKNOWN = 11999 # 未知消息 # 群聊通知类型 MT_CHATROOM_MEMBER_ADD = 11098 # 群成员新增 MT_CHATROOM_MEMBER_REMOVE = 11099 # 群成员删除 MT_CHATROOM_INFO_CHANGE = 11100 # 群信息变化 MT_CHATROOM_MEMBER_NICKNAME_CHANGE = 11101 # 群成员昵称修改 # 发送消息类型 MT_SEND_TEXT = 11036 # 发送文本 # 微信原始消息类型到内部类型的映射 WECHAT_MSG_TYPE_MAP = { "1": MessageType.MT_TEXT, # 文本 "3": MessageType.MT_IMAGE, # 图片 "34": MessageType.MT_VOICE, # 语音 "37": MessageType.MT_FRIEND_REQUEST, # 好友请求 "42": MessageType.MT_CARD, # 名片 "43": MessageType.MT_VIDEO, # 视频 "47": MessageType.MT_EMOJI, # 表情 "48": MessageType.MT_LOCATION, # 位置 "49": MessageType.MT_LINK, # AppMsg，需结合 XML 细分 "10000": MessageType.MT_SYSTEM, # 系统消息 "10002": MessageType.MT_REVOKE, # 撤回消息 } # 消息类型到事件名称的映射 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_CHATROOM_MEMBER_NICKNAME_CHANGE: "chatroom_member_nickname_change", MessageType.MT_LOCATION: "location_message", MessageType.MT_LINK: "link_message", MessageType.MT_FILE: "file_message", MessageType.MT_CARD: "card_message", MessageType.MT_MINIAPP: "miniapp_message", MessageType.MT_UNKNOWN: "other_message", } def _extract_string(value) -> str: """ 提取字符串值 Args: value: 可能是 dict 或 str Returns: 字符串值 """ if isinstance(value, dict): return value.get("String", "") return str(value) if value else "" def _ensure_dict(value) -> dict: """确保返回 dict（兼容字符串/空值）""" if isinstance(value, dict): return value if isinstance(value, str): text = value.strip() if text.startswith("{") or text.startswith("["): try: parsed = json.loads(text) if isinstance(parsed, dict): return parsed except Exception: pass return {} def _strip_group_prefix(content: str) -> str: """去掉群聊消息里可能带的 `wxid:\n` 前缀。""" if not content or not isinstance(content, str): return "" xml_start = content.find(" 0: return content[xml_start:] return content def _parse_appmsg_meta(content) -> dict: """解析 `msgType=49` 的 XML，提取 appmsg 元信息。""" xml_content = _strip_group_prefix(_extract_string(content)) if not xml_content or "<" not in xml_content: return {} try: root = ET.fromstring(xml_content) except Exception: return {} appmsg = root.find(".//appmsg") if appmsg is None and root.tag == "appmsg": appmsg = root if appmsg is None: return {} appattach = appmsg.find("appattach") title = (appmsg.findtext("title", "") or "").strip() desc = (appmsg.findtext("des", "") or appmsg.findtext("description", "") or "").strip() url = (appmsg.findtext("url", "") or "").strip() thumb_url = (appmsg.findtext("thumburl", "") or "").strip() appmsg_type = (appmsg.findtext("type", "") or "").strip() file_name = "" file_ext = "" if appattach is not None: file_name = (appattach.findtext("filename", "") or title).strip() file_ext = (appattach.findtext("fileext", "") or "").strip() return { "xml": xml_content, "appmsg_type": appmsg_type, "title": title, "desc": desc, "url": url, "thumb_url": thumb_url, "has_weappinfo": appmsg.find("weappinfo") is not None, "has_appattach": appattach is not None, "file_name": file_name, "file_ext": file_ext, } def _resolve_appmsg_internal_type(appmsg_meta: dict) -> int: """根据 appmsg 元信息判断 49 消息的真实内部类型。""" appmsg_type = str(appmsg_meta.get("appmsg_type", "")).strip() xml_content = appmsg_meta.get("xml", "") if appmsg_type == "57": return MessageType.MT_QUOTE if appmsg_type in {"33", "36"} or appmsg_meta.get("has_weappinfo"): return MessageType.MT_MINIAPP if appmsg_type == "6" or appmsg_meta.get("has_appattach"): return MessageType.MT_FILE if appmsg_type in {"5", "19"}: return MessageType.MT_LINK if " dict: """ 将新协议的消息格式转换为统一的内部格式 Args: msg_type: 内部消息类型码 data: 原始消息数据（来自 HTTP 回调） Returns: 标准化的消息字典 """ # 判断消息来源类型 message_type_field = data.get("messageType", "") is_group = message_type_field == "群聊消息" # 如果没有 messageType 字段，根据 fromUserName 判断 if not message_type_field: from_user = _extract_string(data.get("fromUserName", {})) is_group = from_user.endswith("@chatroom") # 提取基础字段 from_user = _extract_string(data.get("fromUserName", {})) to_user = _extract_string(data.get("toUserName", {})) content = _extract_string(data.get("content", {})) appmsg_meta = _parse_appmsg_meta(content) if str(data.get("msgType", "")) == "49" else {} # 群聊消息的真实内容 real_content = data.get("real_content", "") if is_group and not real_content: # 从 content 中提取（格式：wxid:\n实际内容） if ":\n" in content: parts = content.split(":\n", 1) if len(parts) == 2: real_content = parts[1] else: real_content = content # 构建标准消息 message = { "MsgType": msg_type, "MsgId": data.get("newMsgId", "") or str(data.get("msgId", "")), "FromWxid": from_user, "ToWxid": to_user, "Content": real_content if is_group else content, "CreateTime": int(data.get("createTime", 0)), "IsGroup": is_group, "RawMsgType": str(data.get("msgType", "1")), } if appmsg_meta: message["AppMsgType"] = appmsg_meta.get("appmsg_type", "") # 群聊消息处理 if is_group: message["RoomWxid"] = from_user # 提取发送者信息 - 优先使用 room_sender_by 字段 room_sender_by = _extract_string(data.get("room_sender_by", "")) member_info = _ensure_dict(data.get("member_info")) if room_sender_by: message["SenderWxid"] = room_sender_by elif member_info: message["SenderWxid"] = member_info.get("userName", "") else: # 从 content 中提取发送者 wxid if ":\n" in content: sender_wxid = content.split(":\n")[0] message["SenderWxid"] = sender_wxid else: message["SenderWxid"] = "" # 发送者昵称 - 从 newChatroomData 中查找 sender_profile = _ensure_dict(data.get("sender_profile")) new_chatroom_data = _ensure_dict(sender_profile.get("newChatroomData")) chatroom_members = new_chatroom_data.get("chatRoomMember") or [] if not isinstance(chatroom_members, list): chatroom_members = [] sender_wxid = message.get("SenderWxid", "") for member in chatroom_members: if member.get("userName") == sender_wxid: message["SenderNickname"] = member.get("nickName", "") break else: if member_info: message["SenderNickname"] = member_info.get("nickName", "") else: message["SenderNickname"] = "" message["SenderAvatar"] = member_info.get("bigHeadImgUrl", "") if member_info else "" else: # 私聊消息 message["SenderWxid"] = from_user message["RoomWxid"] = "" # 提取发送者信息 sender_profile = _ensure_dict(data.get("sender_profile")) if sender_profile: message["SenderNickname"] = _extract_string(sender_profile.get("nickName", {})) message["SenderAvatar"] = sender_profile.get("bigHeadImgUrl", "") else: message["SenderNickname"] = data.get("sender_nick", "") message["SenderAvatar"] = "" # @ 消息处理 content_to_check = real_content if is_group else content if "@" in content_to_check: message["IsAtMessage"] = True # 解析 @ 列表（简单实现） # TODO: 从 msgSource XML 中解析完整的 @ 列表 # 图片消息 if msg_type == MessageType.MT_IMAGE: message["ImgStatus"] = data.get("imgStatus", 0) img_buf = data.get("imgBuf", {}) if img_buf: message["ImgLen"] = img_buf.get("iLen", 0) # 文件消息 if msg_type == MessageType.MT_FILE: message["Filename"] = ( appmsg_meta.get("file_name") or _extract_string(data.get("filename", "")) or appmsg_meta.get("title", "") ) message["FileExtend"] = appmsg_meta.get("file_ext") or _extract_string(data.get("file_extend", "")) # 语音消息 if msg_type == MessageType.MT_VOICE: message["VoiceLength"] = data.get("voiceLength", 0) # 视频消息 if msg_type == MessageType.MT_VIDEO: message["VideoLength"] = data.get("playLength", 0) # 引用消息 if msg_type == MessageType.MT_QUOTE: # 尝试解析引用内容 try: import xml.etree.ElementTree as ET if content: root = ET.fromstring(content) title = root.find(".//title") if title is not None and title.text: message["QuoteTitle"] = title.text refermsg = root.find(".//refermsg") if refermsg is not None: message["QuoteContent"] = refermsg.findtext("content", "") message["QuoteSender"] = refermsg.findtext("fromusr", "") except Exception: pass if msg_type == MessageType.MT_CARD: message["CardWxid"] = _extract_string(data.get("recommend_wxid", "")) message["CardNickname"] = ( _extract_string(data.get("recommend_nickname", "")) or _extract_string(data.get("title", "")) ) # 位置消息 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", "") # 链接消息 if msg_type == MessageType.MT_LINK: message["LinkTitle"] = appmsg_meta.get("title") or _extract_string(data.get("title", "")) message["LinkDesc"] = appmsg_meta.get("desc") or _extract_string(data.get("desc", "")) message["LinkUrl"] = appmsg_meta.get("url") or _extract_string(data.get("url", "")) message["LinkThumb"] = appmsg_meta.get("thumb_url") or _extract_string(data.get("thumb_url", "")) if msg_type == MessageType.MT_MINIAPP: message["MiniAppTitle"] = appmsg_meta.get("title") or _extract_string(data.get("title", "")) message["MiniAppDesc"] = appmsg_meta.get("desc") or _extract_string(data.get("desc", "")) message["MiniAppUrl"] = appmsg_meta.get("url") or _extract_string(data.get("url", "")) message["MiniAppThumb"] = appmsg_meta.get("thumb_url") or _extract_string(data.get("thumb_url", "")) # 好友请求 if msg_type == MessageType.MT_FRIEND_REQUEST: message["V3"] = data.get("v3", "") message["V4"] = data.get("v4", "") message["Scene"] = data.get("scene", 0) # 系统消息 if msg_type == MessageType.MT_SYSTEM: message["Content"] = content # 保留原始数据 message["_raw"] = data return message def get_internal_msg_type(wechat_msg_type: str, data: dict = None) -> int: """ 将微信消息类型转换为内部消息类型 Args: wechat_msg_type: 微信消息类型（字符串） Returns: 内部消息类型码 """ wechat_msg_type = str(wechat_msg_type) if wechat_msg_type == "49" and data is not None: appmsg_meta = _parse_appmsg_meta(data.get("content", "")) if appmsg_meta: return _resolve_appmsg_internal_type(appmsg_meta) return MessageType.MT_LINK return WECHAT_MSG_TYPE_MAP.get(wechat_msg_type, MessageType.MT_UNKNOWN) def normalize_from_callback(message_type: str, data: dict) -> dict: """ 从 HTTP 回调数据标准化消息 Args: message_type: 消息来源类型 (private_message/group_message) data: 原始回调数据 Returns: 标准化的消息字典 """ # 获取微信消息类型 wechat_msg_type = str(data.get("msgType", "1")) # 转换为内部类型 internal_type = get_internal_msg_type(wechat_msg_type, data) # 调用通用标准化函数 return normalize_message(internal_type, data)