消息处理指南
本文说明如何在监听回调中处理笔记、合并消息、小程序、图片/视频/文件等消息。日常开发只需调用 msg.* 高层 API，无需关心内部实现。
阅读顺序建议：
- 先看 消息字段，弄清
attr / type / sender
- 再看 推荐回调模式 和 消息类型 → 处理方法
- 需要解析笔记 / 小程序时，重点看下方「支持的小程序」以及 NoteMessage.get_content
完整可运行示例见 使用示例。
消息字段（attr / type / sender）
每条 Message 对象在 GetNextNewMessage 等 API 的 callback(msg) 或返回列表中可直接读取。
推荐通过 WinAuto.im_client() 绑定当前拾取的主窗口（WIN_CLASS / hwnd），再调用消息 API；默认内置 WeChat 适配器，其它 IM 可传 client='your_pkg.Client'。
用 WeChat() 时，直接 wx.GetNextNewMessage(..., callback=on_message) 即可，字段含义相同。
先按归属过滤、再按类型处理（避免自己给自己自动回复）：
def on_message(msg):
print(f"[{msg.attr}] type={msg.type} sender={msg.sender} content={msg.content}")
if msg.attr == 'friend':
pass # 仅处理对方发来的
elif msg.attr == 'self':
pass # 自己发的，通常跳过自动回复逻辑
elif msg.attr in ('system', 'time'):
pass # 系统/时间行，非聊天内容
GetNextNewMessage 返回值（与 callback 交付的消息一致）：
batch = im.GetNextNewMessage(filter_mute=False, fetch_sender=False, callback=on_message)
# batch = {
# 'chat_name': '文件传输助手', # 当前会话名
# 'chat_type': 'friend', # 'friend' 私聊 | 'group' 群聊
# 'msg': [Message, ...], # 本轮新消息列表
# }
chat = batch.get('chat_name')
chat_type = batch.get('chat_type')
msgs = batch.get('msg') or []
消息锚点（滚动/重绘后找回同一条）
GetNextNewMessage 内部用内容指纹 + runtimeid 标注每条待交付消息；UI 重绘后 msg.id（runtimeid）会变，不能长期依赖 GetMessageById。
anchor = im.MakeMessageAnchor(msg) # callback 内或 GetAllMessage 后
# anchor['key'] / anchor['stable_key'] / anchor['hint_rid'] / anchor['summary']
# 中间滚动、下载其它消息后再找回
msg = im.ResolveMessageByAnchor(anchor)
if msg:
msg.roll_into_view()
msg.download(timeout=15)
详见 消息列表 — MakeMessageAnchor。
完整字段见 msg.info()：
info = msg.info()
# info['attr'], info['type'], info['content'], info['sender'],
# info['quote_nickname'], info['quote_content']（引用消息）, ...
推荐回调模式
下面演示「对方消息」的常见分支：笔记、合并、小程序、图片下载队列。
失败判断： 笔记/小程序/下载失败时，返回值往往是 WxResponse，请用 isinstance(result, WxResponse)，不要写 result or [] 把失败当成空列表。
笔记内容怎么遍历： 成功时是 str（文本）与 Path（本地文件）的混合列表。
from chatautox import WinAuto
from chatautox.param import WxResponse
from pathlib import Path
import time
auto = WinAuto.from_main_hwnd(WIN_CLASS, WIN_TITLES[0], mode='uia')
im = auto.im_client(debug=False, auto_listen=False)
pending_downloads = []
def on_message(msg):
# 先按 attr 过滤，再按 type 分支
if msg.attr != 'friend':
return
if msg.type == 'note':
# wait：笔记窗打开后等待加载完成（多图可设 10）
result = msg.get_content(wait=3)
if isinstance(result, WxResponse):
print('笔记失败:', result.get('message'))
else:
for content in result:
if isinstance(content, str):
print(content)
elif isinstance(content, Path):
print('文件路径：', content)
elif msg.type == 'merge':
for item in msg.get_messages() or []:
print(item)
elif msg.type == 'miniapp':
kind = msg.identify_kind()
if not kind:
print('未内置支持:', msg.content)
return
result = msg.get_content(wait=3)
if isinstance(result, WxResponse):
print('小程序失败:', result.get('message'))
else:
for content in result:
if isinstance(content, str):
print(content)
elif isinstance(content, Path):
print('文件路径：', content)
elif msg.type in ('image', 'video', 'file'):
pending_downloads.append(msg)
def drain_downloads():
while pending_downloads:
msg = pending_downloads.pop(0)
path = msg.download(timeout=15)
if isinstance(path, WxResponse):
print('下载失败', path.get('message'))
else:
print('已保存', path)
while True:
im.GetNextNewMessage(filter_mute=False, callback=on_message)
drain_downloads()
time.sleep(1)
要点：
- 笔记 / 合并 / 小程序：按
msg.type 分支，调用对应方法即可。
- 图片 / 视频 / 文件：建议先入队再批量
download()，避免在 callback 里长时间阻塞导致漏消息。
- 失败判断：用
isinstance(result, WxResponse)，不要只用 or [] 静默跳过。
消息类型 → 处理方法
支持的小程序
凡气泡摘要含「小程序」字样，msg.type 均为 'miniapp'。是否可自动解析由 msg.identify_kind() 决定。
识别优先级（高 → 低）： 微商相册系列 → 产品笔记pro / 私域产品笔记 → 商品笔记。
未内置的小程序
msg.identify_kind() 返回 None- 不会自动点击或下载，仅可读
msg.content 摘要
示例
from chatautox.param import WxResponse
from pathlib import Path
if msg.type == 'miniapp':
if not msg.identify_kind():
print('未内置支持:', msg.content)
else:
result = msg.get_content(wait=3)
if isinstance(result, WxResponse):
print('失败:', result.get('message'))
else:
for content in result:
if isinstance(content, str):
print(content)
elif isinstance(content, Path):
print('文件路径：', content)
与微信原生笔记的区别
二者 type 不同，不会互相冲突。更多参数见 NoteMessage.get_content。
调试环境变量（可选）
更多 API 说明见 Message类 — MiniAppMessage。
消息读取 API 一览
GetNextNewMessage（主路径）
from chatautox import WinAuto
from chatautox.param import WxResponse
auto = WinAuto.from_main_hwnd(WIN_CLASS, WIN_TITLES[0], mode='uia')
im = auto.im_client(debug=False, auto_listen=False)
def on_message(msg):
print(f"[{msg.attr}] {msg.type}: {msg.content}")
batch = im.GetNextNewMessage(
filter_mute=False, # True 跳过免打扰会话
fetch_sender=False, # 群聊 True 补全 msg.sender
callback=on_message, # 每条解析完立刻回调（含 fetch_sender 补全后）
use_profile_sender=False # True 优先资料卡识别发送人
)
print(batch.get('chat_name'), batch.get('chat_type'), len(batch.get('msg') or []))
首帧基线：第一次调用通常返回空 dict（chat_name / msg 为空），仅标记当前视口为已读；持续监听须 while True + time.sleep。
持续监听：
while True:
batch = im.GetNextNewMessage(filter_mute=False, callback=on_message)
if batch.get('msg'):
print(batch['chat_name'], batch['chat_type'], len(batch['msg']))
time.sleep(1)
GetNextUnreadBarMessages（跳转条）
专门处理聊天区「?N条新消息 / N条新消息」浮动条；不会被 GetNextNewMessage 自动处理，需单独调用：
batch = im.GetNextUnreadBarMessages(
count=None,
fetch_sender=False,
callback=on_message,
)
当前窗口：GetAllMessage / GetNewMessage
# 需已 ChatWith 打开目标会话
msgs = im.GetAllMessage(fetch_sender=False)
for msg in msgs:
print(msg.attr, msg.type, msg.sender, msg.content)
new_msgs = im.GetNewMessage() # 仅增量；切换 chat 后首轮为空
子窗口监听
同一会话需先双击弹出独立聊天窗（ChatWnd）：
# 后台线程
im.AddListenChat('文件传输助手', callback=on_message)
im.StartListening()
# 或同步轮询（与 GetNextNewMessage 可同脚本交错）
batch = im.GetNextListenChatMessage(nickname='文件传输助手', callback=on_message)
三种监听方式
同窗口监听需先将会话弹出为独立聊天窗口。
按 type 完整分支示例
参考 测试代码/新消息.py，在 callback 内按类型处理：
from chatautox.param import WxResponse
from pathlib import Path
def on_message(msg):
if msg.attr != 'friend':
return
t = msg.type
if t == 'text':
print('[文本]', msg.content)
elif t == 'quote':
print('[引用]', msg.content)
print(' 被引用:', msg.quote_nickname, '—', msg.quote_content)
elif t == 'voice':
text = msg.to_text() if hasattr(msg, 'to_text') else None
print('[语音]', msg.content, '→', text)
elif t == 'note':
result = msg.get_content(wait=3) # wait：等待笔记加载完成
if isinstance(result, WxResponse):
print('笔记失败:', result.get('message'))
else:
for content in result:
if isinstance(content, str):
print(content)
elif isinstance(content, Path):
print('文件路径：', content)
elif t == 'merge':
for sender, content, _ in msg.get_messages_detailed() or []:
print(sender, content)
elif t in ('image', 'video', 'file'):
path = msg.download(timeout=15)
print('下载:', path)
elif t == 'link':
detail = msg.get_link_detail(timeout=15) if hasattr(msg, 'get_link_detail') else msg.get_url()
print('链接:', detail)
elif t == 'miniapp':
if not msg.identify_kind():
print('未内置小程序:', msg.content)
else:
result = msg.get_content(wait=3)
if isinstance(result, WxResponse):
print('小程序失败:', result.get('message'))
else:
for content in result:
if isinstance(content, str):
print(content)
elif isinstance(content, Path):
print('文件路径：', content)
else:
print(f'[{t}]', msg.content)
合并消息补充
# 纯内容（文本或本地媒体路径字符串）
for item in merge_msg.get_messages():
print(item)
# 含发送者
for sender, content, _ in merge_msg.get_messages_detailed():
print(sender, content)
相关文档：Message类 · WinAuto
