消息处理指南

本文说明如何在监听回调中处理笔记、合并消息、小程序、图片/视频/文件等消息。日常开发只需调用 msg.* 高层 API,无需关心内部实现。


消息字段(attr / type / sender)

每条 Message 对象在 GetNextNewMessage 等 API 的 callback(msg) 或返回列表中可直接读取。
推荐通过 WinAuto.im_client() 绑定当前拾取的主窗口(WIN_CLASS / hwnd),再调用消息 API;默认内置 WeChat 适配器,其它 IM 可传 client='your_pkg.Client'

字段类型说明
msg.attrstr消息归属friend=对方发送,self=自己发送,system=系统提示,time=时间分隔,tickle=拍一拍
msg.typestr消息类型text / image / video / file / voice / note / merge / link / quote / miniapp / location / emotion / personal_card / other
msg.contentstr气泡摘要文字
msg.senderstr发送者昵称(私聊多为 friend/self;群聊需 fetch_sender=True 才补全)
msg.sender_remarkstr备注名(若有)

识别好友发送 vs 自己发送

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 []

群聊补全发送人:fetch_sender=True(默认 OCR;use_profile_sender=True 优先资料卡)。

消息锚点(滚动/重绘后找回同一条)

GetNextNewMessage 内部用内容指纹 + runtimeid 标注每条待交付消息;UI 重绘后 msg.idruntimeid)会变,不能长期依赖 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'](引用消息), ...

推荐回调模式

from chatautox import WinAuto
from chatautox.param import WxResponse
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':
        result = msg.get_content(wait=3)
        if isinstance(result, WxResponse):
            print('笔记失败:', result.get('message'))
        else:
            for item in result:
                print(item)
    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 item in result:
                print(item)
    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推荐调用说明
textmsg.content纯文本
quotemsg.content / msg.quote_nickname / msg.quote_content引用消息;被引用为图片/视频可 download_quote_image(),被引用为文件可 download_quote_file()
voicemsg.to_text()语音转文字
notemsg.get_content(wait=3)读取笔记内容:文本 + 文件路径列表
notemsg.create_note(content, forward_to=None) / wx.create_note(...)创建新笔记;forward_to 为空则不转发(见 NoteMessage#create_note
mergemsg.get_messages()纯内容列表
mergemsg.get_messages_detailed()含发送者:[[sender, content, None], …]
miniapp见下方 支持的小程序内置 3 类可 get_content()
image / video / filemsg.download(timeout=15)建议异步队列下载
linkmsg.content / msg.get_url() / msg.get_link_detail()链接卡片
locationmsg.content / msg.location_data位置
emotionmsg.content表情
personal_cardmsg.content名片
othermsg.content未分类

支持的小程序

凡气泡摘要含「小程序」字样,msg.type 均为 'miniapp'。是否可自动解析由 msg.identify_kind() 决定。

识别优先级(高 → 低): 微商相册系列 → 产品笔记pro / 私域产品笔记 → 商品笔记。

identify_kind() 返回值小程序名称消息摘要关键词(任一命中)用法
weishang_album微商相册、微购相册、种草相册Lite微商相册微购相册种草相册Lite小程序微商相册msg.get_content(wait=3)
product_note_pro产品笔记pro、私域产品笔记产品笔记私域产品笔记小程序产品笔记小程序私域产品笔记同上
product_note商品笔记商品笔记小程序商品笔记(且非 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:
        # get_content 内部会 click 打开小程序并 wait,无需再单独 msg.click()
        result = msg.get_content(wait=3)
        if isinstance(result, WxResponse):
            print('失败:', result.get('message'))
        else:
            for item in result:
                if isinstance(item, str):
                    print('文本:', item)
                elif isinstance(item, Path):
                    print('文件:', item)

与微信原生笔记的区别

微信原生笔记小程序(如商品笔记)
msg.type'note''miniapp'
解析方法msg.get_content(wait=3)msg.identify_kind() + msg.get_content(wait=3)

二者 type 不同,不会互相冲突。

调试环境变量(可选)

变量说明
CHATAUTOX_DEBUG_MINIAPP=1商品笔记 / 产品笔记pro 详细日志
CHATAUTOX_ZIP_TIMEOUT压缩包下载超时(秒,默认 180)
CHATAUTOX_DEBUG_SELECT=0关闭商品笔记「选择下载」流程日志

更多 API 说明见 Message类 — MiniAppMessage


消息读取 API 一览

API场景返回
GetNextNewMessage(...)主窗口会话列表有新消息(推荐dictchat_name / chat_type / msg
GetNextUnreadBarMessages(...)当前聊天内「N条新消息」跳转条同上;无跳转条时 {}
GetAllMessage(fetch_sender=…)当前已打开聊天的全部消息List[Message]
GetNewMessage()当前聊天增量新消息(首帧建立基线)List[Message]
GetNextListenChatMessage(nickname=…)已弹出独立子窗口,同步轮询GetNextNewMessage 结构
AddListenChat(nick, callback) + StartListening()子窗口后台线程监听回调 callback(msg)
ChatInfo()当前会话名与类型dictchat_name / chat_type

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)

三种监听方式

API场景
GetNextNewMessage(callback=…)主窗口会话列表有新消息
GetNextListenChatMessage()已弹出的独立聊天子窗口(同步轮询)
AddListenChat(nick, callback) + StartListening()独立子窗口,后台线程监听

同窗口监听需先将会话弹出为独立聊天窗口。


按 type 完整分支示例

参考 测试代码/新消息.py,在 callback 内按类型处理:

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)
        if isinstance(result, WxResponse):
            print('笔记失败:', result.get('message'))
        else:
            for item in result:
                print(item)
    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)
            print(result)
    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

上一页Message类下一页朋友圈