From f03df5156121eca6b69065c311443c3600b458c9 Mon Sep 17 00:00:00 2001 From: Belugary <53219544+Belugary@users.noreply.github.com> Date: Tue, 12 May 2026 21:03:08 +0800 Subject: [PATCH] feat: parse WeChat transfer messages (appmsg type=2000) (#85) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add structured parsing for transfer messages so they no longer fall through to the generic `[链接/文件]` fallback in chat history exports. Mirrors the dispatch + helper pattern PR #65 (merged-forward type=19) established for `base_type=49` appmsg sub-types. ## What is added **Helpers (mcp_server.py):** - `_TRANSFER_PAYSUBTYPE_LABEL` — maps the 6 community-consensus paysubtypes (1 发起 / 3 已收款 / 4 已退还 / 5 过期已退还 / 7 待领取 / 8 已领取); unknown values degrade to `未知(paysubtype=N)` so a new variant in a future WeChat build is visible rather than silently dropped. - `_extract_transfer_info(appmsg)` — pulls fields out of ``, with snake/camelCase fallback (`feedesc`/`feeDesc`, `pay_memo`/`paymemo`) observed across WeChat versions. - `_format_transfer_message_text(appmsg, title)` — one-line render for chat history: `[转账·已收款] ¥100.00 备注: lunch`. **Dispatch (mcp_server.py):** - `_format_app_message_text` gains an `app_type == 2000` branch that routes to `_format_transfer_message_text`. `get_chat_history`, `export_chat`, `export_all_chats` and `monitor_web` all inherit automatically. **New MCP tool (mcp_server.py):** - `decode_transfer(chat_name, local_id, create_time=0)` — full structured view: direction, amount, memo, payer/receiver wxid, transfer id, transcation id, begin/invalid timestamps. Uses the same multi-shard scan + ambiguity-by-create_time pattern as `decode_file_message` / `decode_record_item`. **CLI wrapper:** - `decode_transfer.py` at the repo root — argparse wrapper that prints the same text as the MCP tool, returning non-zero exit when the message can't be decoded (script-friendly). **JSON export (chat_export_helpers.py + export_chat.py + export_all_chats.py):** - `_extract_content` now returns `(rendered, extras)`. `extras` carries structured fields when a message type has more signal than the human-readable string (currently: transfers → `type:"transfer" + transfer:{direction, fee_desc, pay_memo, ...}`). The channel is forward-compatible — future additions (video号 metadata, expanded merged-forward, etc.) flow through the same shape without changing the caller signature. JSON consumers that only read `content` are unaffected; the change is additive. **monitor_web (monitor_web.py):** - Backend dispatch branch + orange-yellow `.msg-transfer` card CSS + `renderRich` JS handler. ## Tests 12 new cases in `tests/test_record_decoders.py`: - `TransferPaysubTypeLabelTests` — locks the 6-value label table. - `ExtractTransferInfoTests` (6 cases) — full field round-trip, missing `` fallback, snake/camelCase variants, unknown paysubtype degradation, empty paysubtype handling. - `FormatTransferMessageTextTests` (4 cases) — initiate / received-with-memo / missing-wcpayinfo / missing-fee-desc. - `AppMessageDispatchTransferTests` — `_format_app_message_text` routes type=2000 correctly so `get_chat_history` / `export_chat` both pick it up. All fixtures use synthetic placeholder values (`wxid_payer_synth`, `¥100.00`, `1` + 27×`0`); no real PII or transaction IDs. ## Scope 7 files, +546 / -15 (additions only — no behavior change for existing message types). All 180 tests pass locally (168 baseline + 12 new). --- chat_export_helpers.py | 70 +++++++++-- decode_transfer.py | 51 ++++++++ export_all_chats.py | 14 ++- export_chat.py | 16 ++- mcp_server.py | 220 ++++++++++++++++++++++++++++++++++ monitor_web.py | 33 +++++ tests/test_record_decoders.py | 157 ++++++++++++++++++++++++ 7 files changed, 546 insertions(+), 15 deletions(-) create mode 100644 decode_transfer.py diff --git a/chat_export_helpers.py b/chat_export_helpers.py index 2bb3271..5c2794c 100644 --- a/chat_export_helpers.py +++ b/chat_export_helpers.py @@ -113,26 +113,78 @@ def _format_video_message(content): return f"[视频] {playlength}秒" if playlength else "[视频]" +def _extract_transfer_extras(content): + """Detect appmsg type=2000 and return structured transfer fields, else None. + + Reuses mcp_server._extract_transfer_info so the schema/version-quirks logic + lives in one place. Empty values are dropped to keep the export compact. + Numeric timestamps are returned as ints (consistent with the top-level + `timestamp` field), not iso strings — downstream consumers can format. + """ + if not content or ' [] + +参数: + 联系人显示名、备注名或 wxid（仅 1v1 聊天有转账消息）。 + 转账消息的 local_id（从 export_chat 输出 / monitor_web 等地方获取）。 + [] 可选 unix 时间戳。当 local_id 在多个分片冲突时用它唯一定位。 + +输出: 多行可读文本，含方向（发起/收款/退还）、金额、备注、付款/收款 wxid、 + 交易号、发起/失效时间。 + +需先完成 WeChat DB 解密（详见 README）。本 CLI 是 mcp_server.decode_transfer +工具的命令行包装，输出格式与 MCP 工具一致。 +""" +import argparse +import sys + +import mcp_server + + +def main() -> int: + parser = argparse.ArgumentParser( + prog="python3 decode_transfer.py", + description="读取微信转账消息的结构化字段", + ) + parser.add_argument("chat_name", help="联系人名/备注/wxid") + parser.add_argument("local_id", type=int, help="转账消息的 local_id") + parser.add_argument( + "ts", + nargs="?", + type=int, + default=0, + help="消息的 unix 时间戳（跨分片唯一定位时需要，可省略）", + ) + args = parser.parse_args() + + result = mcp_server.decode_transfer(args.chat_name, args.local_id, args.ts) + print(result) + # 如果工具返回错误文案，退出码非 0 便于 shell 脚本判断 + if result.startswith(("错误:", "找不到", "不是转账消息", "无法解析", "消息中没有", "消息 content")): + return 1 + if "无法唯一定位" in result: + return 2 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/export_all_chats.py b/export_all_chats.py index 3e0f99c..cd6e30c 100644 --- a/export_all_chats.py +++ b/export_all_chats.py @@ -62,13 +62,21 @@ def export_one(username, output_dir, names): local_id, local_type, create_time, real_sender_id, content, ct = row sender = _resolve_sender(row, ctx, names, id_to_username) type_str = _msg_type_str(local_type) - rendered = _extract_content(local_id, local_type, content, ct, username, display_name) + rendered, extras = _extract_content( + local_id, local_type, content, ct, username, display_name + ) msg = {"local_id": local_id, "timestamp": create_time, "sender": sender} - if type_str != "text": - msg["type"] = type_str + effective_type = (extras or {}).get("type") or type_str + if effective_type != "text": + msg["type"] = effective_type if rendered is not None: msg["content"] = rendered + if extras: + for k, v in extras.items(): + if k == "type": + continue + msg[k] = v messages.append(msg) if not messages: diff --git a/export_chat.py b/export_chat.py index 8cd8d8f..4c7918f 100644 --- a/export_chat.py +++ b/export_chat.py @@ -85,7 +85,9 @@ def export_chat(chat_name, output_path): local_id, local_type, create_time, real_sender_id, content, ct = row sender = _resolve_sender(row, ctx, names, id_to_username) type_str = _msg_type_str(local_type) - rendered = _extract_content(local_id, local_type, content, ct, username, display_name) + rendered, extras = _extract_content( + local_id, local_type, content, ct, username, display_name + ) # Compact format: omit defaults/nulls. type defaults to "text", transcription # is added later by transcribe_chat.py only for voice messages. See CLAUDE.md. @@ -94,10 +96,18 @@ def export_chat(chat_name, output_path): "timestamp": create_time, "sender": sender, } - if type_str != "text": - msg["type"] = type_str + # extras may override type with a more specific value (e.g. "transfer" + # narrower than the generic "link_or_file" base=49 maps to). + effective_type = (extras or {}).get("type") or type_str + if effective_type != "text": + msg["type"] = effective_type if rendered is not None: msg["content"] = rendered + if extras: + for k, v in extras.items(): + if k == "type": + continue + msg[k] = v messages.append(msg) output = { diff --git a/mcp_server.py b/mcp_server.py index 456b448..1077baa 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -732,6 +732,9 @@ def _format_app_message_text(content, local_type, is_group, chat_username, chat_ if app_type == 19: return _format_record_message_text(appmsg, title) + if app_type == 2000: + return _format_transfer_message_text(appmsg, title) + if app_type == 6: return f"[文件] {title}" if title else "[文件]" if app_type == 5: @@ -845,6 +848,80 @@ def _format_record_message_text(appmsg, title): return "\n".join(lines) +# 微信转账 (appmsg type=2000, ) paysubtype 含义。 +# 微信官方无公开文档，此表来自社区抓包归纳。1/3/4 在所有已知版本一致； +# 5/7/8 在不同版本存在变体（"过期已退还"在某些抓包里也归为 4），所以遇到 +# 未识别值时降级显示原始数字，方便用户自行核对。 +_TRANSFER_PAYSUBTYPE_LABEL = { + '1': '发起转账', # 发送方记录：等待对方收钱 + '3': '已收款', # 双向：发送方看到"对方已收"，接收方看到"已收钱" + '4': '已退还', # 主动退还或被退还 + '5': '过期已退还', # 24h 未收，自动退还（发送方记录） + '7': '待领取', # 已发起未接收 + '8': '已领取', # 部分版本：转账被领取（接收方记录） +} + + +def _extract_transfer_info(appmsg): + """从 appmsg type=2000 解出 wcpayinfo 各字段，返回 dict 或 None。 + + 字段大小写在不同微信版本间漂移（见过 feedesc/feeDesc, pay_memo/paymemo）， + 用 lower-case 兜底。所有值用 _collapse_text 清掉换行/前后空白。 + """ + info = appmsg.find('wcpayinfo') + if info is None: + return None + + def _pick(*tags): + for t in tags: + v = _collapse_text(info.findtext(t) or '') + if v: + return v + return '' + + paysubtype = _pick('paysubtype') + return { + 'paysubtype': paysubtype, + 'paysubtype_label': _TRANSFER_PAYSUBTYPE_LABEL.get( + paysubtype, f'未知(paysubtype={paysubtype})' if paysubtype else '' + ), + # feedesc 通常是 "¥0.01" 风格的展示串；feedescxml 是富文本变体 + 'fee_desc': _pick('feedesc', 'feeDesc'), + 'pay_memo': _pick('pay_memo', 'paymemo'), + # 三种交易号：transcationid 是微信支付侧（注意拼写是 transc 不是 trans）， + # transferid 是微信内部转账 id，paymsgid 偶见于旧版本 + 'transcation_id': _pick('transcationid', 'transcationId'), + 'transfer_id': _pick('transferid', 'transferId'), + 'pay_msg_id': _pick('paymsgid', 'payMsgId'), + 'begin_transfer_time': _pick('begintransfertime', 'beginTransferTime'), + 'invalid_time': _pick('invalidtime', 'invalidTime'), + 'effective_date': _pick('effectivedate', 'effectiveDate'), + 'payer_username': _pick('payer_username', 'payerUsername'), + 'receiver_username': _pick('receiver_username', 'receiverUsername'), + } + + +def _format_transfer_message_text(appmsg, title): + """渲染微信转账（appmsg type=2000）一行展示文本，给 history / monitor_web 共用。 + + fallback 顺序： + 1) wcpayinfo 缺失 → 只显示 title 兜底，避免吞数据 + 2) paysubtype 未知 → 显示原始数字让用户自查 + 3) 没有 fee_desc → 至少给个方向标签 + """ + info = _extract_transfer_info(appmsg) + if not info: + return f"[转账] {title}" if title else "[转账]" + + label = info['paysubtype_label'] or '转账' + parts = [f"[转账·{label}]"] if label != '转账' else ["[转账]"] + if info['fee_desc']: + parts.append(info['fee_desc']) + if info['pay_memo']: + parts.append(f"备注: {info['pay_memo']}") + return ' '.join(parts) + + def _format_voip_message_text(content): if not content or ' str: + """读取微信转账消息（appmsg type=2000）的结构化信息。 + + 返回方向（发起/收款/退还）、金额、备注、付款人/收款人 wxid、交易号、 + 发起/失效时间。仅 1v1 聊天有转账消息（微信不支持群转账）。 + + 使用流程：先用 get_chat_history 找到 [转账·xxx] 行 (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}" + + # 多分片扫描 + ambiguity 检测，跟 decode_file_message 一致 + 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_transfer(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=2000" + ) + + 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 != 2000: + return ( + f"不是转账消息（appmsg type={app_type}）。" + f"转账要求 appmsg type=2000；type=6 是文件，type=19 是合并转发，" + f"请用对应的 decode_file_message / decode_record_item 工具" + ) + + info = _extract_transfer_info(appmsg) + if info is None: + return "消息是 type=2000 但缺 节点（schema 异常）" + + 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})' + + direction = info['paysubtype_label'] or '(未知)' + raw_paysubtype = info['paysubtype'] or '?' + title = _collapse_text(appmsg.findtext('title') or '') or '微信转账' + des = _collapse_text(appmsg.findtext('des') or '') + + lines = [f"转账消息: {title}"] + if des: + lines.append(f" 描述: {des}") + lines.append(f" 方向: {direction} (paysubtype={raw_paysubtype})") + if info['fee_desc']: + lines.append(f" 金额: {info['fee_desc']}") + if info['pay_memo']: + lines.append(f" 备注: {info['pay_memo']}") + if info['payer_username']: + lines.append(f" 付款方 wxid: {info['payer_username']}") + if info['receiver_username']: + lines.append(f" 收款方 wxid: {info['receiver_username']}") + begin_ts = _fmt_ts(info['begin_transfer_time']) + if begin_ts: + lines.append(f" 发起时间: {begin_ts}") + invalid_ts = _fmt_ts(info['invalid_time']) + if invalid_ts: + lines.append(f" 失效时间: {invalid_ts}") + if info['transfer_id']: + lines.append(f" 转账 ID: {info['transfer_id']}") + if info['transcation_id']: + lines.append(f" 支付交易号: {info['transcation_id']}") + if info['pay_msg_id']: + lines.append(f" paymsgid: {info['pay_msg_id']}") + return "\n".join(lines) + + @mcp.tool() def get_chat_images(chat_name: str, limit: int = 20) -> str: """列出某个聊天中的图片消息。 diff --git a/monitor_web.py b/monitor_web.py index b9cb1a9..77feeb6 100644 --- a/monitor_web.py +++ b/monitor_web.py @@ -1317,6 +1317,29 @@ class SessionMonitor: 'des': des[:200] if des else '', 'items': items, } + elif app_type == 2000: + # 微信转账 — paysubtype 含义为社区共识表，1/3/4 跨版本一致； + # 字段名在不同版本有 snake/camel 漂移，逐个尝试 + info = appmsg.find('wcpayinfo') + paysubtype = '' + fee_desc = '' + pay_memo = '' + if info is not None: + paysubtype = (info.findtext('paysubtype') or '').strip() + fee_desc = (info.findtext('feedesc') or info.findtext('feeDesc') or '').strip() + pay_memo = (info.findtext('pay_memo') or info.findtext('paymemo') or '').strip() + direction = { + '1': '发起转账', '3': '已收款', '4': '已退还', + '5': '过期已退还', '7': '待领取', '8': '已领取', + }.get(paysubtype, '') + return { + 'type': 'transfer', + 'title': title or '微信转账', + 'direction': direction, + 'paysubtype': paysubtype, + 'fee_desc': fee_desc, + 'pay_memo': pay_memo[:200] if pay_memo else '', + } else: # 其他子类型: 用 title 显示 if title: @@ -1657,6 +1680,10 @@ body{font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,sans-serif;b .chatlog-item{font-size:12px;color:#999;line-height:1.5;white-space:nowrap;overflow:hidden;text-overflow:ellipsis} .chatlog-item b{color:#bbb;font-weight:500} .chatlog-more{font-size:11px;color:#555;margin-top:4px} +.msg-transfer{display:inline-block;background:rgba(255,170,60,.1);border:1px solid rgba(255,170,60,.25);border-radius:8px;padding:8px 14px;margin-top:4px;min-width:180px} +.msg-transfer-head{font-size:13px;color:#ffb84d;font-weight:500} +.msg-transfer-amount{font-size:18px;color:#ffd28a;font-weight:600;margin-top:4px} +.msg-transfer-memo{font-size:11px;color:#999;margin-top:4px} a.msg-link{text-decoration:none;color:inherit} #lightbox{display:none;position:fixed;top:0;left:0;width:100%;height:100%;background:rgba(0,0,0,.92);z-index:1000;cursor:zoom-out;justify-content:center;align-items:center} #lightbox.show{display:flex} @@ -1772,6 +1799,12 @@ function renderRich(r){ } return `
📋 ${esc(r.title)}
${body}
`; } + if(r.type==='transfer') { + let dirLabel = r.direction || '微信转账'; + let amount = r.fee_desc ? '
'+esc(r.fee_desc)+'
' : ''; + let memo = r.pay_memo ? '
备注: '+esc(r.pay_memo)+'
' : ''; + return `
💸 ${esc(dirLabel)}
${amount}${memo}
`; + } if(r.type==='voice') return `
🎤 语音 ${r.duration}s
`; if(r.type==='video') return `
🎬 视频${r.duration?' '+r.duration+'s':''}
`; return null; diff --git a/tests/test_record_decoders.py b/tests/test_record_decoders.py index ef2dfd2..0221b95 100644 --- a/tests/test_record_decoders.py +++ b/tests/test_record_decoders.py @@ -309,5 +309,162 @@ class FormatRecordMessageTextTests(unittest.TestCase): mcp_server._RECORD_MAX_ITEMS = original_max +# -------- WCPay transfer (appmsg type=2000) -------------------------------- +# +# All fixtures use synthetic placeholder values — no real wxid / fee / id / +# memo. paysubtype semantics are community consensus from open-source wechat +# tooling; treat any "未识别" branch as forward-compatible degradation. + + +class TransferPaysubTypeLabelTests(unittest.TestCase): + def test_known_subtypes_present(self): + labels = mcp_server._TRANSFER_PAYSUBTYPE_LABEL + self.assertEqual(labels['1'], '发起转账') + self.assertEqual(labels['3'], '已收款') + self.assertEqual(labels['4'], '已退还') + # 5/7/8 are version-dependent variants — locked to current text so a + # silent rename in mcp_server.py would surface here. + self.assertEqual(labels['5'], '过期已退还') + self.assertEqual(labels['7'], '待领取') + self.assertEqual(labels['8'], '已领取') + + +def _transfer_appmsg( + paysubtype='1', + fee_desc='¥100.00', + pay_memo='', + payer='wxid_payer_synth', + receiver='wxid_recv_synth', + transferid='1' + '0' * 27, + transcationid='1' + '0' * 27, + begin_ts='1746528000', + invalid_ts='1746614400', + title='微信转账', + des='请收钱', + feedesc_tag='feedesc', + paymemo_tag='pay_memo', +): + """Build a synthetic appmsg type=2000 root element. All values are + placeholder; tests never run against real wechat data.""" + import xml.etree.ElementTree as ET + fee_node = f'<{feedesc_tag}>{fee_desc}' if fee_desc else '' + memo_node = f'<{paymemo_tag}>{pay_memo}' if pay_memo else '' + xml_text = ( + f'{title}{des}' + f'2000' + f'' + f'{paysubtype}' + f'{fee_node}{memo_node}' + f'{transferid}' + f'{transcationid}' + f'{begin_ts}' + f'{invalid_ts}' + f'{payer}' + f'{receiver}' + f'' + ) + return ET.fromstring(xml_text), xml_text + + +class ExtractTransferInfoTests(unittest.TestCase): + def test_full_fields_round_trip(self): + root, _ = _transfer_appmsg(paysubtype='3', pay_memo='lunch split') + appmsg = root.find('.//appmsg') + info = mcp_server._extract_transfer_info(appmsg) + self.assertIsNotNone(info) + self.assertEqual(info['paysubtype'], '3') + self.assertEqual(info['paysubtype_label'], '已收款') + self.assertEqual(info['fee_desc'], '¥100.00') + self.assertEqual(info['pay_memo'], 'lunch split') + self.assertEqual(info['payer_username'], 'wxid_payer_synth') + self.assertEqual(info['receiver_username'], 'wxid_recv_synth') + self.assertEqual(info['begin_transfer_time'], '1746528000') + self.assertEqual(info['invalid_time'], '1746614400') + self.assertTrue(info['transfer_id'].startswith('1')) + self.assertTrue(info['transcation_id'].startswith('1')) + + def test_missing_wcpayinfo_returns_none(self): + import xml.etree.ElementTree as ET + root = ET.fromstring( + 'x2000' + ) + appmsg = root.find('.//appmsg') + self.assertIsNone(mcp_server._extract_transfer_info(appmsg)) + + def test_camelcase_feedesc_falls_back(self): + # 部分微信版本字段名为 feeDesc 而非 feedesc + root, _ = _transfer_appmsg(feedesc_tag='feeDesc') + appmsg = root.find('.//appmsg') + info = mcp_server._extract_transfer_info(appmsg) + self.assertEqual(info['fee_desc'], '¥100.00') + + def test_camelcase_paymemo_falls_back(self): + # paymemo (无下划线) 也是已知变体 + root, _ = _transfer_appmsg(pay_memo='note', paymemo_tag='paymemo') + appmsg = root.find('.//appmsg') + info = mcp_server._extract_transfer_info(appmsg) + self.assertEqual(info['pay_memo'], 'note') + + def test_unknown_paysubtype_label_degraded(self): + root, _ = _transfer_appmsg(paysubtype='99') + appmsg = root.find('.//appmsg') + info = mcp_server._extract_transfer_info(appmsg) + self.assertEqual(info['paysubtype'], '99') + self.assertIn('99', info['paysubtype_label']) + + def test_empty_paysubtype_label_empty(self): + root, _ = _transfer_appmsg(paysubtype='') + appmsg = root.find('.//appmsg') + info = mcp_server._extract_transfer_info(appmsg) + self.assertEqual(info['paysubtype_label'], '') + + +class FormatTransferMessageTextTests(unittest.TestCase): + def test_initiate_with_amount(self): + root, _ = _transfer_appmsg(paysubtype='1') + appmsg = root.find('.//appmsg') + out = mcp_server._format_transfer_message_text(appmsg, '微信转账') + self.assertIn('[转账·发起转账]', out) + self.assertIn('¥100.00', out) + + def test_received_with_memo(self): + root, _ = _transfer_appmsg(paysubtype='3', pay_memo='lunch') + appmsg = root.find('.//appmsg') + out = mcp_server._format_transfer_message_text(appmsg, '微信转账') + self.assertIn('[转账·已收款]', out) + self.assertIn('备注: lunch', out) + + def test_missing_wcpayinfo_falls_back_to_title(self): + import xml.etree.ElementTree as ET + root = ET.fromstring( + '微信转账2000' + ) + appmsg = root.find('.//appmsg') + out = mcp_server._format_transfer_message_text(appmsg, '微信转账') + self.assertEqual(out, '[转账] 微信转账') + + def test_missing_fee_desc_safe(self): + # 没有金额时也要给一行能看的输出，不能崩 + root, _ = _transfer_appmsg(paysubtype='4', fee_desc='') + appmsg = root.find('.//appmsg') + out = mcp_server._format_transfer_message_text(appmsg, '微信转账') + self.assertIn('[转账·已退还]', out) + + +class AppMessageDispatchTransferTests(unittest.TestCase): + """type=2000 must route through _format_transfer_message_text via + _format_app_message_text (so get_chat_history / export_chat both pick it up).""" + + def test_dispatch_calls_transfer_helper(self): + _, xml_text = _transfer_appmsg(paysubtype='3', pay_memo='dinner') + out = mcp_server._format_app_message_text( + xml_text, 49, False, 'wxid_dummy', 'dummy', {} + ) + self.assertIsNotNone(out) + self.assertIn('[转账·已收款]', out) + self.assertIn('¥100.00', out) + self.assertIn('备注: dinner', out) + + if __name__ == '__main__': unittest.main()