消息监听

pywxauto 提供基于事件驱动的消息监听机制，通过装饰器注册事件处理器，自动识别消息类型并分发。

基本用法

python

from pywxauto import Weixin, Event

wx = Weixin()

# 注册文本消息处理器
@wx.on(Event.TEXT)
def on_text(weixin, chat, message):
    print(f"{message.sender}: {message.content}")

# 注册要监听的会话（会打开为独立窗口）
wx.add_chat_listen(["张三", "工作群"])

# 启动监听（阻塞，Ctrl+C 退出）
wx.run()

事件类型

事件	说明
`Event.ALL`	所有消息
`Event.TEXT`	文本消息
`Event.IMAGE`	图片消息
`Event.VIDEO`	视频消息
`Event.FILE`	文件消息
`Event.VOICE`	语音消息
`Event.QUOTE`	引用消息
`Event.EMOTION`	表情消息
`Event.LOCATION`	位置消息
`Event.LINK`	链接消息
`Event.CARD`	卡片消息
`Event.PERSONAL_CARD`	个人名片
`Event.MERGE`	合并消息
`Event.NOTE`	笔记消息
`Event.VOIP`	通话消息
`Event.RED_PACKET`	红包消息
`Event.TRANSFER`	转账消息
`Event.MUSIC`	音乐消息
`Event.SYSTEM`	系统消息
`Event.OTHER`	其他消息

监听多种消息类型

python

@wx.on([Event.TEXT, Event.IMAGE, Event.FILE])
def on_message(weixin, chat, message):
    print(f"[{message.type_label}] {message.sender}: {message.content}")

监听所有消息

python

@wx.on(Event.ALL)
def on_all(weixin, chat, message):
    print(message)

一次性监听

python

@wx.once(Event.TEXT)
def on_first_text(weixin, chat, message):
    print("收到第一条文本:", message.content)
    # 触发一次后自动移除

回调参数

所有回调函数的签名为：

python

def handler(weixin: Weixin, chat: SeparateChat, message: Message):
    ...

参数	类型	说明
`weixin`	`Weixin`	微信实例
`chat`	`SeparateChat`	消息来源的独立聊天窗口
`message`	`Message`	消息对象（具体子类）

在回调中回复

python

@wx.on(Event.TEXT)
def auto_reply(weixin, chat, message):
    if message.source.value == "others":
        chat.send_text(f"自动回复: 收到你的消息 [{message.content}]")

管理监听会话

python

# 添加监听
wx.add_chat_listen(["张三", "李四"])

# 自动发现所有已打开的独立窗口
wx.add_all_chats_listen()

# 移除监听
wx.remove_chat_listen(["张三"])

# 移除所有监听
wx.remove_all_chats_listen()

自动扫描新窗口

python

# auto_scan=True: 自动发现并监听新打开的独立聊天窗口
wx.run(auto_scan=True)

停止监听

python

import threading

# 从其他线程停止
def stop_after_10s():
    import time
    time.sleep(10)
    wx.stop()

threading.Thread(target=stop_after_10s, daemon=True).start()
wx.run()

移除事件处理器

python

# 移除特定处理器
wx.off(Event.TEXT, on_text)

# 移除某事件的所有处理器
wx.off(Event.TEXT)

# 移除所有处理器
wx.off()

轮询参数

python

wx.run(
    interval=0.1,       # 有新消息时的轮询间隔（秒）
    idle_interval=0.1,  # 无新消息时的轮询间隔（秒）
    auto_scan=False,    # 是否自动扫描新窗口
    listen_mode="ui",   # 监听模式: "ui"（UI 自动化）或 "db"（数据库监听）
)

数据库监听模式

除了默认的 UI 模式（通过独立聊天窗口监听），pywxauto 还支持数据库监听模式——直接监控微信的加密 SQLite 数据库文件，无需打开任何聊天窗口。

优势

无需打开独立聊天窗口，自动监听所有会话
不依赖窗口状态，更加稳定
资源消耗低，通过 WAL 文件变化触发扫描

基本用法

python

from pywxauto import Weixin, Event

# 需要指定微信进程 PID
wx = Weixin(pid=12345)

@wx.on(Event.TEXT)
def on_text(weixin, chat, message):
    # 数据库模式下 chat 为 None
    # message 是一个字典
    print(f"{message['sender']}: {message['msg']}")

@wx.on(Event.ALL)
def on_all(weixin, chat, message):
    print(f"[{message['type']}] {message['from_wxid']} -> {message['to_wxid']}")

# 启动数据库监听
wx.run(listen_mode="db")

回调参数差异

数据库模式的回调签名与 UI 模式相同，但参数内容不同：

参数	UI 模式	DB 模式
`weixin`	`Weixin` 实例	`Weixin` 实例
`chat`	`SeparateChat` 对象	`None`
`message`	`Message` 对象	`dict` 字典

message 字典字段

字段	类型	说明
`database`	`str`	数据库文件名
`table`	`str`	消息表名
`id`	`int`	本地消息 ID（local_id）
`msg_id`	`int`	服务器消息 ID（server_id）
`sequence`	`int`	排序序列号（sort_seq）
`type`	`int`	消息类型（对应 DBEvent 枚举值）
`is_sender`	`int`	是否为自己发送（1=是, 0=否）
`msg`	`str \| dict`	消息内容（文本为 str，非文本消息解析为 dict）
`source`	`dict \| None`	消息来源 XML 解析结果
`at_user_list`	`list`	@用户列表
`from_wxid`	`str \| None`	发送者 wxid
`sender`	`str \| None`	发送者昵称
`to_wxid`	`str \| None`	接收者 wxid
`receiver`	`str \| None`	接收者昵称
`room_wxid`	`str \| None`	群聊 wxid（群消息时有值）
`room_nickname`	`str \| None`	群聊昵称
`extra`	`bytes \| None`	附加信息（packed_info_data）
`status`	`int`	消息状态
`create_time`	`int`	消息创建时间戳

DBEvent 枚举类型

数据库模式中 message['type'] 对应以下枚举值：

枚举值	整数值	说明	映射到 Event
`DBEvent.TEXT`	1	文本消息	`Event.TEXT`
`DBEvent.TEXT2`	2	文本消息（变体）	`Event.TEXT`
`DBEvent.IMAGE`	3	图片消息	`Event.IMAGE`
`DBEvent.VOICE`	34	语音消息	`Event.VOICE`
`DBEvent.CARD`	42	名片消息	`Event.CARD`
`DBEvent.VIDEO`	43	视频消息	`Event.VIDEO`
`DBEvent.EMOTION`	47	表情消息	`Event.EMOTION`
`DBEvent.LOCATION`	48	位置消息	`Event.LOCATION`
`DBEvent.VOIP`	50	通话消息	`Event.VOIP`
`DBEvent.OPEN_IM_CARD`	66	企业名片	`Event.CARD`
`DBEvent.SYSTEM`	10000	系统消息	`Event.SYSTEM`
`DBEvent.FILE`	25769803825	文件消息	`Event.FILE`
`DBEvent.FILE_WAIT`	317827579953	文件待接收	`Event.FILE`
`DBEvent.LINK`	21474836529	链接消息	`Event.LINK`
`DBEvent.LINK2`	292057776177	链接消息（变体）	`Event.LINK`
`DBEvent.SONG`	12884901937	音乐消息	`Event.MUSIC`
`DBEvent.RED_ENVELOPE`	8594229559345	红包消息	`Event.RED_PACKET`
`DBEvent.TRANSFER`	8589934592049	转账消息	`Event.TRANSFER`
`DBEvent.QUOTE`	244813135921	引用消息	`Event.QUOTE`
`DBEvent.MERGED_FORWARD`	81604378673	合并转发	`Event.MERGE`
`DBEvent.FINDER_VIDEO`	219043332145	视频号视频	`Event.OTHER`
`DBEvent.COLLECTION`	103079215153	收藏消息	`Event.OTHER`
`DBEvent.PAT`	266287972401	拍一拍	`Event.OTHER`
`DBEvent.GROUP_ANNOUNCEMENT`	373662154801	群公告	`Event.SYSTEM`

完整示例

python

from pywxauto import Weixin, Event

wx = Weixin(pid=12345)

@wx.on(Event.TEXT)
def on_text(weixin, chat, message):
    sender = message["sender"]
    content = message["msg"]
    room = message.get("room_nickname")
    if room:
        print(f"[{room}] {sender}: {content}")
    else:
        print(f"{sender}: {content}")

@wx.on(Event.IMAGE)
def on_image(weixin, chat, message):
    print(f"收到图片消息来自: {message['sender']}")

@wx.on(Event.FILE)
def on_file(weixin, chat, message):
    print(f"收到文件消息来自: {message['sender']}")

@wx.on(Event.ALL)
def on_all(weixin, chat, message):
    print(f"[{message['database']}/{message['table']}] "
          f"type={message['type']} "
          f"{message['from_wxid']} -> {message['to_wxid'] or message['room_wxid']}")

wx.run(listen_mode="db")

注意事项

需要指定 pid 参数（用于读取进程内存获取数据库密钥）
需要管理员权限运行（读取进程内存）
依赖 pymem、sqlcipher3、zstandard、xmltodict（已包含在安装依赖中）
微信进程需要运行中且已登录
该模式不支持发送消息，仅用于接收和处理消息
非文本消息的 msg 字段会被自动解析为 XML 字典

WeixinManager 多实例数据库监听

使用 WeixinManager 可以同时监听多个微信客户端的数据库：

python

from pywxauto import WeixinManager, Event

wx = WeixinManager()

@wx.on(Event.TEXT)
def on_text(weixin, chat, message):
    print(f"[PID={weixin.pid}] {message['sender']}: {message['msg']}")

@wx.on(Event.ALL)
def on_all(weixin, chat, message):
    print(f"[PID={weixin.pid}] [{message['type']}] {message['from_wxid']}")

# 所有已连接客户端同时启动数据库监听
wx.run(listen_mode="db")

完整示例：智能机器人

python

from pywxauto import Weixin, Event

wx = Weixin(background=True)

@wx.on(Event.TEXT)
def on_text(weixin, chat, message):
    if message.source.value != "others":
        return
    content = message.content.strip()
    if content == "你好":
        chat.send_text("你好！有什么可以帮你的？")
    elif content == "时间":
        import datetime
        chat.send_text(f"现在是 {datetime.datetime.now().strftime('%H:%M:%S')}")

@wx.on(Event.IMAGE)
def on_image(weixin, chat, message):
    if message.source.value != "others":
        return
    chat.send_text("收到图片！")

@wx.on(Event.FILE)
def on_file(weixin, chat, message):
    if message.source.value != "others":
        return
    chat.send_text(f"收到文件: {message.file_name} ({message.file_size})")

wx.add_chat_listen(["文件传输助手"])
wx.run()

消息监听 ​

基本用法 ​

事件类型 ​

监听多种消息类型 ​

监听所有消息 ​

一次性监听 ​

回调参数 ​

在回调中回复 ​

管理监听会话 ​

自动扫描新窗口 ​

停止监听 ​

移除事件处理器 ​

轮询参数 ​

数据库监听模式 ​

优势 ​

基本用法 ​

回调参数差异 ​

message 字典字段 ​

DBEvent 枚举类型 ​

完整示例 ​

注意事项 ​

WeixinManager 多实例数据库监听 ​

完整示例：智能机器人 ​

消息监听

基本用法

事件类型

监听多种消息类型

监听所有消息

一次性监听

回调参数

在回调中回复

管理监听会话

自动扫描新窗口

停止监听

移除事件处理器

轮询参数

数据库监听模式

优势

基本用法

回调参数差异

message 字典字段

DBEvent 枚举类型

完整示例

注意事项

WeixinManager 多实例数据库监听

完整示例：智能机器人