From a6cb3d04973c6851710d131c6fde70368bcccd60 Mon Sep 17 00:00:00 2001 From: Belugary <53219544+Belugary@users.noreply.github.com> Date: Wed, 13 May 2026 13:05:21 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=A7=A3=E6=9E=90=E5=BE=AE=E4=BF=A1?= =?UTF-8?q?=E5=BC=95=E7=94=A8=E5=9B=9E=E5=A4=8D=E6=B6=88=E6=81=AF=20(appms?= =?UTF-8?q?g=20type=3D57)=20+=20=E6=96=B0=E5=A2=9E=20decode=5Frefer=20MCP?= =?UTF-8?q?=20=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit > 高价值改动 rationale (override 路径) > > 引用回复 (appmsg type=57) 是聊天里第 3 高频的消息类型 (仅次于纯文本和 > 图片)。当前 _format_app_message_text 的 type=57 分支直接把 refermsg/ > content 按 [:160] 截断当摘要,对内层 type=3 (图片) / 34 (语音) / > 43 (视频) / 47 (动画表情) / 49 (嵌套卡片) 这些"二进制"被引用消息, > 会把 cdnurl / aeskey / md5 / cdnthumb / voiceurl / externurl 一坨乱码 > 渲染到 LLM 可见的 chat history,严重污染上下文。issue #44 #45 重复反馈 > 一个月无人接 —— 这是个明确的用户痛点,fork 实测覆盖 5 种内层 type 的真 > 实数据,渲染长度从原本几千字降到 21-58 字。改动较大但 review 风险低: > 替换的就是 19 行 inline 截断逻辑,新加的 helpers / decode_refer 都是 > 纯加,不动现有 API。 \`_format_app_message_text\` 当前 type=57 分支用 19 行 inline 逻辑直接 \`refer.findtext('content')[:160]\` 当摘要。这对 type=1 (文本) 工作正常, 但对其他内层 type 是个隐藏的 bug: - type=3 图片: 渲染 \`\` 截断 - type=34 语音: 渲染 \`\` 截断 - type=43 视频: 渲染 \`\` 截断 - type=47 动画表情: 渲染 \`\` 截断 - type=49 嵌套卡片: 渲染外层 escape 后的 XML 字符串截断 后果: cdnurl / aeskey / md5 / voiceurl / externurl 等二进制元数据泄漏到 LLM 可见的聊天历史,污染上下文且无信息量。引用回复是 type=57 是高频消息,影响面大。 按 refer_type 分发 schema-aware 摘要: 1. **新增三组 helpers (mcp_server.py +135 行,纯加)**: - \`_REFER_INNER_TYPE_LABEL\`: 内层 type → 中文标签 (1 文本 / 3 图片 / 34 语音 / ...) - \`_INNER_APPMSG_TYPE_LABEL\`: refer_type=49 时嵌套 appmsg/type → 标签 (5 链接 / 6 文件 / 19 聊天记录 / ...) - \`_extract_refer_info(appmsg)\`: 提取 refermsg 全字段返回 dict - \`_summarize_refer_content(refer_type, content)\`: 按 type 分支 - type=1: 取原文,截断到 max_len - type=3/34/43/47/...: 给标签,**不**展开 cdnurl/aeskey/md5 - type=49: 走 \`_parse_xml_root\` (经 \`_XML_UNSAFE_RE\` 过滤 DOCTYPE/ENTITY 防 XXE) 解一层 inner appmsg, 给 \`[链接] xxx\` - 未识别 type: 给 \`[type=N]\` 兜底 - \`_format_refer_message_text(appmsg, ...)\`: 渲染两行格式 \`<回复正文>\n ↳ 回复 <对方>: <摘要>\` 2. **\`_format_app_message_text\` 的 type=57 分支简化**: 19 行 inline → 3 行 dispatch 到 helper。 3. **新增 MCP 工具 \`decode_refer(chat_name, local_id, create_time=0)\`**: 输出结构化多行文本 (回复正文 / 被引用发送者 / 类型 / 摘要 / svrid / createtime), 错误文案分别指引 \`decode_file_message\` (type=6) / \`decode_record_item\` (type=19) / \`decode_transfer\` (type=2000), 不让用户在 4 个工具间盲猜。 新文件 \`tests/test_refer_message.py\`, 20 个新测试: - \`ReferInnerTypeLabelTests\` (2): 标签映射 spot-check - \`ExtractReferInfoTests\` (2): 全字段提取 / refermsg 缺失返回 None - \`SummarizeReferContentTests\` (11): 5 种 refer_type 标签 / type=1 文本截断 / type=49 嵌套链接卡 / type=49 聊天记录卡 / type=49 invalid XML 退化 / unknown type 兜底 / 空 content / XXE payload 拒绝 - \`FormatReferMessageTextTests\` (4): 1v1 文本引用渲染 / 图片引用不泄漏 PII (cdnurl/aeskey/md5) / refermsg 缺失退回 title / 空 reply 用占位符 - \`AppMessageDispatchReferTests\` (1): dispatcher 走新 helper 不走旧截断 合成 fixture (wxid_synth_a/b, 12345@chatroom, Sender A/B, svrid 1+0\*18), 无真实 PII。 基线 183 → 203 通过 (+20 新增), 0 回归。 - \`mcp_server.py\`: 替换 19 行 type=57 inline → 3 行 dispatch (净 -16 行); 新增 6 个 helpers + 1 个 MCP 工具 \`decode_refer\` (+275 行); 不改任何现有公开 API。 - \`tests/test_refer_message.py\`: 新增 (20 测试, 合成 fixture, 不依赖真实加密素材)。 - **本 PR 不包含 fork 里的 CLI 入口 (\`wxdec.cli.decode_refer\`) 和 \`export_chat\` / \`monitor_web\` 的对应改动** —— 那几处依赖 fork 私有的包结构 (\`wxdec/cli/\`), 不属于上游 scope。后续如有需要可单独提。 issue #44 #45 (引用回复渲染乱码) --- mcp_server.py | 296 +++++++++++++++++++++++++++++++++--- tests/test_refer_message.py | 219 ++++++++++++++++++++++++++ 2 files changed, 496 insertions(+), 19 deletions(-) create mode 100644 tests/test_refer_message.py diff --git a/mcp_server.py b/mcp_server.py index 8c6ca45..bc360ba 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -731,25 +731,9 @@ def _format_app_message_text(content, local_type, is_group, chat_username, chat_ app_type = _parse_int(app_type_text, _parse_int(sub_type, 0)) if app_type == 57: - ref = appmsg.find('.//refermsg') - ref_user = '' - ref_display_name = '' - ref_content = '' - if ref is not None: - ref_user = (ref.findtext('fromusr') or '').strip() - ref_display_name = (ref.findtext('displayname') or '').strip() - ref_content = _collapse_text(ref.findtext('content') or '') - if len(ref_content) > 160: - ref_content = ref_content[:160] + "..." - - quote_text = title or "[引用消息]" - if ref_content: - ref_label = _resolve_quote_sender_label( - ref_user, ref_display_name, is_group, chat_username, chat_display_name, names - ) - prefix = f"回复 {ref_label}: " if ref_label else "回复: " - quote_text += f"\n ↳ {prefix}{ref_content}" - return quote_text + return _format_refer_message_text( + appmsg, is_group, chat_username, chat_display_name, names + ) if app_type == 19: return _format_record_message_text(appmsg, title) @@ -884,6 +868,135 @@ _TRANSFER_PAYSUBTYPE_LABEL = { } +# 微信引用回复（appmsg type=57, ）内层 的标签映射。 +# refermsg/ 用的是顶层 base_type 数字（跟 format_msg_type 重合）， +# 但语义不同：format_msg_type 给"消息类型 chip"，这里给"被引用消息的一行摘要"， +# 不展开 cdn url / aeskey / md5 等二进制元数据（直接截断 XML 字符串当摘要是 +# 现状的 bug，会把"图片/语音/视频/动画表情/嵌套卡片"渲染成乱码——见 issue #44 #45）。 +_REFER_INNER_TYPE_LABEL = { + '1': '文本', # 特殊：直接展开 content + '3': '图片', + '34': '语音', + '42': '名片', + '43': '视频', + '47': '动画表情', + '48': '位置', + '49': '链接/卡片', # 特殊：嵌套 appmsg，进一步解 inner type + '50': '通话', +} + +# refer_type=49 时 content 是嵌套 ...，inner appmsg/ → 标签。 +# 跟合并转发 _RECORD_DATATYPE_LABEL 的数字含义不同（datatype 是 recorditem 的私有 +# schema），独立维护。 +_INNER_APPMSG_TYPE_LABEL = { + '5': '链接', '6': '文件', '8': '动画表情卡', + '19': '聊天记录', '33': '小程序', '36': '小程序', + '51': '视频号', '57': '引用消息', + '2000': '转账', '2001': '红包', +} + + +def _extract_refer_info(appmsg): + """从 appmsg type=57 解出 refermsg 各字段，返回 dict 或 None。 + + refermsg/ 是 escape 后的字符串，内层 type 决定其 schema: + type=1 (纯文本) / 3 (img cdn) / 34 (voicemsg) / 47 (emoji) + / 49 (嵌套 appmsg) / ... + + refer_content 保留原始字符串（不 collapse），让 _summarize_refer_content + 按 type 进一步处理（type=49 还要再解一层 XML）。其他字段过 _collapse_text + 清掉换行/前后空白。 + """ + refer = appmsg.find('refermsg') + if refer is None: + return None + + return { + 'reply_text': _collapse_text(appmsg.findtext('title') or ''), + 'refer_type': _collapse_text(refer.findtext('type') or ''), + 'refer_svrid': _collapse_text(refer.findtext('svrid') or ''), + 'refer_fromusr': _collapse_text(refer.findtext('fromusr') or ''), + 'refer_chatusr': _collapse_text(refer.findtext('chatusr') or ''), + 'refer_displayname': _collapse_text(refer.findtext('displayname') or ''), + 'refer_content': refer.findtext('content') or '', + 'refer_createtime': _collapse_text(refer.findtext('createtime') or ''), + } + + +def _summarize_refer_content(refer_type, content, max_len=160): + """把被引用消息的 content 摘要成一行可读文本。 + + 分支规则： + type=1 (文本): 取原文，截断到 max_len + type=3/34/43/47/...: 给标签兜底，不展开 cdn url / aeskey / md5 + type=49 (嵌套 appmsg): 解一层 inner appmsg/type + title，给"[链接] xxx" + 未识别 type: 给 [type=N] 兜底，方便用户自查 + + max_len 只对 type=1 文本生效；标签型摘要本身就短。 + """ + refer_type = (refer_type or '').strip() + + if not content: + label = _REFER_INNER_TYPE_LABEL.get(refer_type) + if label: + return f'[{label}]' + return f'[type={refer_type}]' if refer_type else '[引用消息]' + + if refer_type == '1': + text = _collapse_text(content) + return text[:max_len] + '…' if len(text) > max_len else text + + if refer_type == '49': + # 嵌套 appmsg：content 是来源不可信的微信侧 payload，走 _parse_xml_root + # 经 _XML_UNSAFE_RE 过滤 DOCTYPE/ENTITY 防 XXE 注入。 + inner_root = _parse_xml_root(content) + if inner_root is None: + return '[卡片]' + inner_appmsg = inner_root.find('.//appmsg') + if inner_appmsg is None: + return '[卡片]' + inner_type = _collapse_text(inner_appmsg.findtext('type') or '') + inner_title = _collapse_text(inner_appmsg.findtext('title') or '') + label = _INNER_APPMSG_TYPE_LABEL.get( + inner_type, f'卡片 type={inner_type}' if inner_type else '卡片' + ) + return f'[{label}] {inner_title}' if inner_title else f'[{label}]' + + label = _REFER_INNER_TYPE_LABEL.get(refer_type) + if label: + return f'[{label}]' + return f'[type={refer_type}]' + + +def _format_refer_message_text(appmsg, is_group, chat_username, chat_display_name, names): + """渲染微信引用回复（appmsg type=57）的两行展示文本。 + + 格式: + <用户的回复正文> + ↳ 回复 <对方>: <被引用消息摘要> + + fallback: + 1) refermsg 缺失 → 退回到外层 title 兜底 + 2) refer_content 空 → summary 给"[refer_type 标签]"或"[引用消息]" + 3) sender 解析不出来 → "回复:" 不带名字 + """ + info = _extract_refer_info(appmsg) + if info is None: + title = _collapse_text(appmsg.findtext('title') or '') + return title or '[引用消息]' + + summary = _summarize_refer_content(info['refer_type'], info['refer_content']) + sender_label = _resolve_quote_sender_label( + info['refer_fromusr'], info['refer_displayname'], + is_group, chat_username, chat_display_name, names + ) + + quote_text = info['reply_text'] or '[引用消息]' + prefix = f'回复 {sender_label}: ' if sender_label else '回复: ' + quote_text += f'\n ↳ {prefix}{summary}' + return quote_text + + def _extract_transfer_info(appmsg): """从 appmsg type=2000 解出 wcpayinfo 各字段，返回 dict 或 None。 @@ -2829,6 +2942,151 @@ def decode_transfer(chat_name: str, local_id: int, create_time: int = 0) -> str: return "\n".join(lines) +@mcp.tool() +def decode_refer(chat_name: str, local_id: int, create_time: int = 0) -> str: + """读取微信引用回复消息（appmsg type=57）的结构化信息。 + + 返回回复正文、被引用消息的发送者/类型/摘要/svrid/createtime。被引用消息的 + type 决定摘要风格：1 文本展开原文，3/34/43/47/48/50 给 [图片]/[语音]/... + 标签，49 嵌套 appmsg 解一层 inner type 给 [链接] xxx。svrid 可用于回查 + 原消息（在 history / export_chat 输出里搜）。 + + 使用流程：先用 get_chat_history 找到 [引用消息] 行 (local_id=N, ts=T)， + 把 N 和 T 一起传进来。create_time(ts) 用于跨分片场景下唯一定位。 + + Args: + chat_name: 聊天对象的名字、备注名或 wxid + local_id: 引用消息的 local_id（从 get_chat_history 获取） + create_time: 消息的 unix 时间戳，从 get_chat_history 输出 ts=N 部分获取。 + 用于在 local_id 跨分片冲突时唯一定位；传 0 时若多个分片含同 local_id 会报歧义错误 + """ + try: + local_id = int(local_id) + create_time = int(create_time) + except (TypeError, ValueError): + return "错误: local_id 和 create_time 必须是整数" + + username = resolve_username(chat_name) + if not username: + return f"找不到聊天对象: {chat_name}" + + shards = _find_msg_tables_for_user(username) + if not shards: + return f"找不到 {chat_name} 的消息表" + + matches = [] + for shard in shards: + if not _is_safe_msg_table_name(shard['table_name']): + continue + with closing(sqlite3.connect(shard['db_path'])) as conn: + if create_time: + candidate_row = conn.execute( + f"SELECT local_type, create_time, message_content, WCDB_CT_message_content " + f"FROM [{shard['table_name']}] WHERE local_id=? AND create_time=?", + (local_id, create_time) + ).fetchone() + else: + candidate_row = conn.execute( + f"SELECT local_type, create_time, message_content, WCDB_CT_message_content " + f"FROM [{shard['table_name']}] WHERE local_id=?", + (local_id,) + ).fetchone() + if candidate_row: + matches.append((shard['db_path'], candidate_row)) + + if not matches: + if create_time: + return f"找不到 (local_id={local_id}, create_time={create_time}) 的消息（已扫描 {len(shards)} 个分片）" + return f"找不到 local_id={local_id} 的消息（已扫描 {len(shards)} 个分片）" + if len(matches) > 1: + details = [] + for db_p, r in matches: + ct = r[1] + ts_str = datetime.fromtimestamp(ct).isoformat() if ct else '?' + details.append(f"{os.path.basename(db_p)} create_time={ct} ({ts_str})") + return ( + f"local_id={local_id} 在 {len(matches)} 个分片中都存在，无法唯一定位:\n " + + '\n '.join(details) + + f"\n请加 create_time 参数：decode_refer(chat_name, local_id={local_id}, create_time=N)" + ) + + _, row = matches[0] + local_type, msg_create_time, content, ct_compress = row + base_type, _ = _split_msg_type(local_type) + if base_type != 49: + return ( + f"不是引用消息（local_type={local_type}, base_type={base_type}），" + f"引用消息应为 base_type=49 + appmsg type=57" + ) + + xml_text = _decompress_content(content, ct_compress) + if not xml_text: + return "消息 content 为空或无法解码" + + is_group = username.endswith('@chatroom') + _, xml_text = _parse_message_content(xml_text, local_type, is_group) + + root = _parse_app_message_outer(xml_text) + if root is None: + return "无法解析消息 XML" + appmsg = root.find('.//appmsg') + if appmsg is None: + return "消息中没有 appmsg 段（不像引用回复）" + + app_type = _parse_int(_collapse_text(appmsg.findtext('type') or ''), 0) + if app_type != 57: + return ( + f"不是引用消息（appmsg type={app_type}）。" + f"引用回复要求 appmsg type=57；type=6 是文件、type=19 是合并转发、" + f"type=2000 是转账，请用对应的 decode_file_message / decode_record_item / " + f"decode_transfer 工具" + ) + + info = _extract_refer_info(appmsg) + if info is None: + return "消息是 type=57 但缺 节点（schema 异常）" + + refer_type_label = _REFER_INNER_TYPE_LABEL.get(info['refer_type'], '') + summary = _summarize_refer_content(info['refer_type'], info['refer_content']) + sender_label = _resolve_quote_sender_label( + info['refer_fromusr'], info['refer_displayname'], + is_group, username, chat_name, get_contact_names() + ) + + def _fmt_ts(ts_str): + ts = _parse_int(ts_str, 0) + if not ts: + return '' + try: + return datetime.fromtimestamp(ts).isoformat() + except (ValueError, OSError, OverflowError): + return f'(无效 ts={ts_str})' + + lines = [f"引用回复消息: {info['reply_text'] or '(无回复正文)'}"] + if sender_label: + lines.append(f" 被引用消息发送者: {sender_label}") + if info['refer_displayname']: + lines.append(f" 被引用消息显示名: {info['refer_displayname']}") + if info['refer_fromusr']: + lines.append(f" 被引用消息 from: {info['refer_fromusr']}") + if info['refer_chatusr']: + lines.append(f" 被引用消息 chatusr (群内发送者 wxid): {info['refer_chatusr']}") + raw_type = info['refer_type'] or '?' + type_display = ( + f"{refer_type_label} (refer_type={raw_type})" + if refer_type_label else f"refer_type={raw_type}" + ) + lines.append(f" 被引用消息类型: {type_display}") + lines.append(f" 被引用消息摘要: {summary}") + refer_ts = _fmt_ts(info['refer_createtime']) + if refer_ts: + lines.append(f" 被引用消息创建时间: {refer_ts}") + if info['refer_svrid']: + lines.append(f" 被引用消息 server_id: {info['refer_svrid']}") + + return "\n".join(lines) + + @mcp.tool() def get_chat_images(chat_name: str, limit: int = 20, offset: int = 0, start_time: str = "", end_time: str = "") -> str: """列出某个聊天中的图片消息。 diff --git a/tests/test_refer_message.py b/tests/test_refer_message.py new file mode 100644 index 0000000..fc763f5 --- /dev/null +++ b/tests/test_refer_message.py @@ -0,0 +1,219 @@ +"""微信引用回复消息（appmsg type=57）解析鉴定测试。 + +旧逻辑直接把 refermsg/content 按 [:160] 截断当摘要，对 type=3 (图片) / +34 (语音) / 47 (动画表情) / 49 (嵌套卡片) 这些"二进制"被引用消息会渲染 +成 cdnurl + aeskey + md5 一坨乱码 (issue #44 #45)。本组测试 pin 新行为： +按 refer_type 给 schema-aware 摘要，cdnurl / aeskey / md5 / cdnthumb / +voiceurl / externurl 全部不再泄漏到聊天历史。 + +合成 fixture：wxid_synth_a / wxid_synth_b / 12345@chatroom / Sender A/B / +svrid 1 + 0*18，无真实 PII。 +""" +import unittest +import xml.etree.ElementTree as ET + +import mcp_server + + +# ---------- 合成 fixture ---------- + +def _appmsg(refermsg_xml='', title='我的回复'): + """组装一个最小 type=57 appmsg 元素。""" + xml = ( + f'57{title}' + f'{refermsg_xml}' + ) + root = ET.fromstring(xml) + return root.find('.//appmsg') + + +def _refermsg(refer_type, content, fromusr='wxid_synth_a', + displayname='Sender A', svrid='1' + '0' * 18, + chatusr='', createtime='1700000000'): + return ( + '' + f'{refer_type}' + f'{svrid}' + f'{fromusr}' + f'{chatusr}' + f'{displayname}' + f'{createtime}' + f'{content}' + '' + ) + + +# ---------- 标签映射 ---------- + +class ReferInnerTypeLabelTests(unittest.TestCase): + def test_known_refer_inner_labels(self): + self.assertEqual(mcp_server._REFER_INNER_TYPE_LABEL['3'], '图片') + self.assertEqual(mcp_server._REFER_INNER_TYPE_LABEL['34'], '语音') + self.assertEqual(mcp_server._REFER_INNER_TYPE_LABEL['47'], '动画表情') + self.assertEqual(mcp_server._REFER_INNER_TYPE_LABEL['49'], '链接/卡片') + + def test_known_inner_appmsg_labels(self): + self.assertEqual(mcp_server._INNER_APPMSG_TYPE_LABEL['5'], '链接') + self.assertEqual(mcp_server._INNER_APPMSG_TYPE_LABEL['6'], '文件') + self.assertEqual(mcp_server._INNER_APPMSG_TYPE_LABEL['19'], '聊天记录') + + +# ---------- _extract_refer_info ---------- + +class ExtractReferInfoTests(unittest.TestCase): + def test_full_fields_round_trip(self): + appmsg = _appmsg(_refermsg('1', '原文本'), title='回复正文') + info = mcp_server._extract_refer_info(appmsg) + self.assertEqual(info['reply_text'], '回复正文') + self.assertEqual(info['refer_type'], '1') + self.assertEqual(info['refer_fromusr'], 'wxid_synth_a') + self.assertEqual(info['refer_displayname'], 'Sender A') + self.assertEqual(info['refer_svrid'], '1' + '0' * 18) + self.assertEqual(info['refer_content'], '原文本') + + def test_missing_refermsg_returns_none(self): + appmsg = _appmsg(refermsg_xml='', title='孤儿回复') + self.assertIsNone(mcp_server._extract_refer_info(appmsg)) + + +# ---------- _summarize_refer_content ---------- + +class SummarizeReferContentTests(unittest.TestCase): + def test_text_returns_original(self): + self.assertEqual(mcp_server._summarize_refer_content('1', '你好'), '你好') + + def test_text_truncates_to_max_len(self): + long = '中' * 200 + out = mcp_server._summarize_refer_content('1', long, max_len=160) + self.assertEqual(len(out), 161) # 160 + '…' + self.assertTrue(out.endswith('…')) + + def test_image_returns_label_not_xml(self): + v2_image_xml = ( + '' + ) + out = mcp_server._summarize_refer_content('3', v2_image_xml) + self.assertEqual(out, '[图片]') + # PII / 二进制元数据不能泄漏到摘要 + for leak in ('cdnurl', 'aeskey', 'md5', 'cdnthumb', 'leak_main'): + self.assertNotIn(leak, out) + + def test_voice_returns_label(self): + v_xml = '' + out = mcp_server._summarize_refer_content('34', v_xml) + self.assertEqual(out, '[语音]') + self.assertNotIn('voiceurl', out) + + def test_emoji_returns_label(self): + out = mcp_server._summarize_refer_content( + '47', '' + ) + self.assertEqual(out, '[动画表情]') + self.assertNotIn('externurl', out) + self.assertNotIn('leak', out) + + def test_nested_link_card_summary(self): + nested = '5分享标题'\ + 'http://example.com/leak' + out = mcp_server._summarize_refer_content('49', nested) + self.assertEqual(out, '[链接] 分享标题') + self.assertNotIn('http', out) + self.assertNotIn('url', out) + + def test_nested_record_card_summary(self): + nested = '19群聊天记录' + out = mcp_server._summarize_refer_content('49', nested) + self.assertEqual(out, '[聊天记录] 群聊天记录') + + def test_nested_invalid_xml_falls_back_to_card(self): + self.assertEqual( + mcp_server._summarize_refer_content('49', ']>' + '5&x;' + ) + out = mcp_server._summarize_refer_content('49', xxe) + self.assertEqual(out, '[卡片]') + + +# ---------- _format_refer_message_text ---------- + +class FormatReferMessageTextTests(unittest.TestCase): + def _names(self): + return {'wxid_synth_a': 'Sender A', 'wxid_synth_b': 'Sender B'} + + def test_text_refer_in_1v1(self): + appmsg = _appmsg(_refermsg('1', '你吃了吗'), title='吃了') + out = mcp_server._format_refer_message_text( + appmsg, is_group=False, chat_username='wxid_synth_a', + chat_display_name='Sender A', names=self._names(), + ) + self.assertEqual(out, '吃了\n ↳ 回复 Sender A: 你吃了吗') + + def test_image_refer_uses_label_not_xml_payload(self): + v2_image = ( + '<msg><img cdnurl="leak" aeskey="leak" md5="leak"/></msg>' + ) + appmsg = _appmsg(_refermsg('3', v2_image), title='这张?') + out = mcp_server._format_refer_message_text( + appmsg, is_group=False, chat_username='wxid_synth_a', + chat_display_name='Sender A', names=self._names(), + ) + self.assertIn('[图片]', out) + for leak in ('cdnurl', 'aeskey', 'md5'): + self.assertNotIn(leak, out) + + def test_missing_refermsg_falls_back_to_title(self): + appmsg = _appmsg(refermsg_xml='', title='孤儿回复') + out = mcp_server._format_refer_message_text( + appmsg, is_group=False, chat_username='wxid_synth_a', + chat_display_name='Sender A', names={}, + ) + self.assertEqual(out, '孤儿回复') + + def test_empty_reply_uses_placeholder(self): + appmsg = _appmsg(_refermsg('1', 'hi'), title='') + out = mcp_server._format_refer_message_text( + appmsg, is_group=False, chat_username='wxid_synth_a', + chat_display_name='Sender A', names=self._names(), + ) + self.assertTrue(out.startswith('[引用消息]')) + + +# ---------- 调度入口 ---------- + +class AppMessageDispatchReferTests(unittest.TestCase): + def test_type57_dispatches_to_helper(self): + # _format_app_message_text 的 type=57 分支必须走 _format_refer_message_text， + # 不再走旧的 inline [:160] 截断。 + v2_image = '<msg><img cdnurl="leak_main"/></msg>' + content = ( + f'57看这个' + f'{_refermsg("3", v2_image)}' + ) + out = mcp_server._format_app_message_text( + content, local_type=49, is_group=False, + chat_username='wxid_synth_a', chat_display_name='Sender A', names={}, + ) + self.assertIn('[图片]', out) + self.assertNotIn('leak_main', out) + self.assertNotIn('cdnurl', out) + + +if __name__ == '__main__': + unittest.main()