微信防撤回消息记录机器人

前言

微信消息撤回功能让很多用户苦不堪言——刚看到一半的合同截图、客户投诉记录、群里的爆料截图，转眼间就消失了。对于企业运营、客服团队和个人用户来说，能够自动拦截并保存撤回消息是一个高频刚需。本文详细介绍如何基于个人微信API 搭建一套真正稳定可用的防撤回消息记录机器人，覆盖原理、接入流程、核心代码和运营注意事项。

一、防撤回的技术原理：从协议层拦截才是正解

为什么客户端插件不可靠

市面上流传的"防撤回"方案大致分两类：

PC 客户端 Hook 注入：通过修改微信 Windows 客户端内存或 DLL，拦截撤回指令。这类方案随微信更新频繁失效，且容易触发微信风控。
协议层监听：在消息下发阶段就把原始消息持久化，撤回通知到来时对比本地记录——这才是工程上真正可靠的方式。

微信iPad协议方案属于后者。iPad 协议是微信官方为 iPad 客户端开放的同步协议，消息时序完整、字段齐全，可以稳定区分"普通消息"和"撤回通知"两种消息类型，是搭建防撤回机器人的最优底层协议。

撤回消息的协议特征

在 iPad 协议的消息推送流中，每条消息有独立的 msgId 和 createTime。当发送方撤回一条消息时，微信服务器会下发一条特殊的系统消息，其 msgType 为 10002（系统通知类型），内容为 XML 格式，核心结构如下：

xml<sysmsg type="revokemsg">
  <revokemsg>
    <session>wxid_xxxxxxxxxxxxxxx</session>
    <oldmsgid>1234567890</oldmsgid>
    <msgid>9876543210</msgid>
    <replacemsg><![CDATA["xxx" 撤回了一条消息]]></replacemsg>
  </revokemsg>
</sysmsg>

其中 oldmsgid 是原始消息的 ID，session 是会话（联系人或群）的 wxid。机器人只需要在收到此类通知时，根据 oldmsgid 查本地缓存，就能把原消息内容重新推送给指定目标。

二、整体架构设计

一套完整的防撤回机器人包含以下模块：

模块	职责	技术选型建议
消息接收层	接收 WechatApi 的 Webhook 回调推送	Python Flask / Node.js Express
消息缓存层	持久化原始消息（含图片/文件下载链接）	Redis（设 TTL 24h）或 SQLite
撤回检测层	解析 `msgType=10002` 的 XML，提取 `oldmsgid`	xml.etree / fast-xml-parser
消息重发层	调用 WechatApi 发送接口，将原消息转发回目标会话	HTTP POST
管理后台（可选）	查看撤回记录、设置白名单群/联系人	任意 Web 框架

整个系统以 WechatApi 为核心引擎——设备登录、消息收发、媒体文件下载全部通过其 HTTP 接口完成，业务逻辑只需关注撤回检测和转发策略，开发成本极低。

三、接入 WechatApi：设备登录与 Webhook 配置

3.1 注册并获取 appId

前往 WechatApi 控制台注册账号，创建设备后会得到：

appId：设备唯一标识，每个接入的微信号对应一个 appId
VideosApi-token：API 鉴权 Token，所有请求必须在 Header 中携带

3.2 配置消息回调

在控制台的"回调设置"页面，填入你的服务器公网地址，例如：

http://your-server.com:8080/wechat/callback

WechatApi 会将该微信号收到的每一条消息实时 POST 到这个地址，Body 为 JSON 格式。一条普通文本消息的回调示例：

json{
  "appId": "wx_device_abcd1234",
  "msgId": "1234567890123",
  "fromUser": "wxid_sender001",
  "toUser": "wxid_receiver001",
  "msgType": 1,
  "content": "这条消息稍后会被撤回",
  "createTime": 1718000000,
  "isChatRoom": false,
  "chatRoomId": ""
}

一条撤回通知的回调示例（msgType 为 10002）：

json{
  "appId": "wx_device_abcd1234",
  "msgId": "9876543210987",
  "fromUser": "wxid_sender001",
  "toUser": "wxid_receiver001",
  "msgType": 10002,
  "content": "<sysmsg type=\"revokemsg\"><revokemsg><session>wxid_receiver001</session><oldmsgid>1234567890123</oldmsgid><msgid>9876543210987</msgid><replacemsg><![CDATA[\"xxx\" 撤回了一条消息]]></replacemsg></revokemsg></sysmsg>",
  "createTime": 1718000060,
  "isChatRoom": false,
  "chatRoomId": ""
}

四、核心代码实现：Python 版本

以下是一个最小可运行的防撤回机器人核心逻辑，基于 Python + Flask + Redis：

pythonimport json
import redis
import requests
import xml.etree.ElementTree as ET
from flask import Flask, request

app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

WECHAT_API_BASE = "https://api.wechatapi.net"  # 示意，请以实际文档为准
APP_ID = "wx_device_abcd1234"
API_TOKEN = "your_videos_api_token_here"

HEADERS = {
    "VideosApi-token": API_TOKEN,
    "Content-Type": "application/json"
}


def cache_message(msg: dict):
    """将收到的消息缓存到 Redis，TTL 设为 24 小时"""
    key = f"msg:{msg['appId']}:{msg['msgId']}"
    r.setex(key, 86400, json.dumps(msg, ensure_ascii=False))


def get_cached_message(app_id: str, msg_id: str):
    """根据 appId + msgId 查询缓存"""
    key = f"msg:{app_id}:{msg_id}"
    val = r.get(key)
    return json.loads(val) if val else None


def parse_revoke_msgid(content: str) -> str | None:
    """从撤回通知 XML 中提取原始 msgId"""
    try:
        root = ET.fromstring(content)
        revoke = root.find('revokemsg')
        if revoke is not None:
            return revoke.findtext('oldmsgid')
    except ET.ParseError:
        pass
    return None


def send_text(to_user: str, text: str):
    """调用 WechatApi 发送文本消息"""
    payload = {
        "appId": APP_ID,
        "toUser": to_user,
        "content": text
    }
    resp = requests.post(
        f"{WECHAT_API_BASE}/message/sendText",
        headers=HEADERS,
        json=payload,
        timeout=10
    )
    return resp.json()


@app.route('/wechat/callback', methods=['POST'])
def callback():
    msg = request.get_json(force=True)

    if msg.get('msgType') == 10002:
        # 撤回通知：解析原始 msgId
        old_msg_id = parse_revoke_msgid(msg.get('content', ''))
        if old_msg_id:
            original = get_cached_message(msg['appId'], old_msg_id)
            if original:
                # 确定转发目标：发给撤回操作触发的会话
                target = msg.get('chatRoomId') or msg.get('fromUser')
                summary = (
                    f"[防撤回] {original.get('fromUser')} 刚才撤回了一条消息：\n"
                    f"{original.get('content', '[非文本消息]')}"
                )
                send_text(target, summary)
    else:
        # 普通消息：缓存起来备查
        cache_message(msg)

    return {"ret": 200, "msg": "ok"}


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

发送接口的标准返回格式

WechatApi 所有接口统一返回如下结构：

json{
  "ret": 200,
  "msg": "success",
  "data": {
    "msgId": "9999999999001",
    "createTime": 1718000120
  }
}

ret 为 200 表示成功，其他值（如 400、401、500）配合 msg 字段提示具体错误原因。建议在业务代码里统一判断 ret == 200，而非依赖 HTTP 状态码。

五、图片和文件类撤回消息的处理

文本消息的防撤回相对简单，难点在于图片、语音、视频、文件等媒体消息。收到媒体消息的回调时，content 字段通常是 XML，包含媒体资源的 aeskey、cdnurl 等字段。

推荐的处理策略：

收到媒体消息时：立即调用 WechatApi 的媒体下载接口，将资源下载到本地或对象存储（如七牛、阿里云 OSS），并把本地路径一并缓存到 Redis。
检测到撤回时：取出本地路径，调用 WechatApi 的发送图片/文件接口重新发送。

bash# 示意：调用发送图片接口（curl 版本）
curl -X POST https://api.wechatapi.net/message/sendImage \
  -H "VideosApi-token: your_videos_api_token_here" \
  -H "Content-Type: application/json" \
  -d '{
    "appId": "wx_device_abcd1234",
    "toUser": "wxid_target001",
    "imageUrl": "https://your-oss.com/revoke_backup/img_1234567890.jpg"
  }'

对于语音消息，由于微信语音使用私有格式（SILK），需要额外做格式转换才能重发，这部分可以借助 silk-v3-decoder 库处理。

六、群聊防撤回的特殊处理

群聊场景比单聊更复杂，需要注意以下几点：

6.1 区分是群成员撤回还是群主/管理员撤回

撤回通知中的 fromUser 在群聊场景下是群的 wxid（即 chatRoomId），而实际操作撤回的人记录在 content XML 的 replacemsg 字段文本中。如果需要精确记录"谁撤回了"，需要额外解析这段文本或结合群成员昵称列表做映射。

6.2 设置监听白名单/黑名单

不建议对所有群都开启防撤回推送，否则会产生大量噪音。可以在 Redis 或配置文件里维护一个群 ID 白名单，只对列表内的群处理撤回事件。

6.3 转发目标的选择

防撤回消息可以有几种转发策略：

转发策略	适用场景	说明
原群内转发	家庭群、业务沟通群	所有成员都能看到被撤回内容
转发到私人存档群	客服团队、审计需求	创建一个只有管理员的私密群，撤回内容静默存入
转发给文件传输助手	个人用户	最低调，只有自己能看
写入数据库，仅 Web 后台可查	合规审计场景	不在微信内二次传播

七、稳定性与合规注意事项

7.1 消息缓存的 TTL 设置

Redis 的 TTL 建议设为 24 小时。微信消息撤回有时间限制（通常 2 分钟内），但为了保险，将缓存时间拉长到 24 小时可以覆盖几乎所有撤回场景，同时避免缓存无限膨胀。如果消息量非常大（如监听几十个高活跃群），可以改用 SQLite 或 MySQL 替代 Redis，并定期清理 3 天以前的记录。

7.2 设备稳定性

采用微信二次开发方案时，设备的长期在线是关键。WechatApi 基于 iPad 协议，设备掉线后需要重新登录，期间的消息会丢失。建议：

配置心跳检测接口，定时检查设备在线状态
掉线后自动触发重新扫码登录流程，并通过告警通知运维人员
不要在同一个微信号上同时运行多个设备实例，否则会互踢

7.3 消息内容的隐私合规

防撤回本质上是对他人消息的二次保存和转发，在企业场景中使用前应当：

明确告知群成员该群已开启消息留存
遵守《个人信息保护法》等相关法规，不对敏感消息（如身份证、银行卡图片）进行自动化处理
仅在授权范围内使用，不得用于恶意监控或数据倒卖

7.4 异常处理与幂等性

回调服务需要做好幂等处理——WechatApi 在网络抖动时可能重发相同的回调，导致同一条撤回消息被推送两次。可以用 Redis 的 SETNX 对已处理的 msgId 做去重：

pythondef is_processed(msg_id: str) -> bool:
    key = f"processed:{msg_id}"
    # SETNX 返回 True 表示首次处理，False 表示已处理过
    return not r.set(key, 1, nx=True, ex=3600)

小结

搭建微信防撤回消息记录机器人的核心思路是：在消息下发时缓存，在撤回通知到来时查缓存并重发。整个流程中最重要的基础设施是一个稳定可靠的微信消息收发 API——WechatApi 基于 iPad 协议，提供完整的消息回调、媒体下载和发送能力，是实现这一方案的推荐选择。从单聊文本防撤回到群聊媒体防撤回，再到合规审计存档，本文覆盖的技术方案均可在此基础上快速落地。如需进一步了解接口细节，可访问 WechatApi 开发文档或前往控制台申请试用设备。