Files
zWorkFlow/decode_transfer.py
Belugary f03df51561 feat: parse WeChat transfer messages (appmsg type=2000) (#85)
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 `<wcpayinfo>`,
  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
  `<wcpayinfo>` 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).
2026-05-12 21:03:08 +08:00

52 lines
1.7 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
读取微信转账消息appmsg type=2000的结构化字段。
用法:
python3 decode_transfer.py <chat_name> <local_id> [<ts>]
参数:
<chat_name> 联系人显示名、备注名或 wxid仅 1v1 聊天有转账消息)。
<local_id> 转账消息的 local_id从 export_chat 输出 / monitor_web 等地方获取)。
[<ts>] 可选 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())