微信聊天记录自动备份机器人

前言

微信是国内最主流的即时通讯工具，大量业务沟通、客户对话、合同协商都发生在微信里。然而微信官方不提供聊天记录云备份接口，手机换机或账号被封时历史消息极易永久丢失。更严重的是，企业在合规审计、纠纷举证等场景下，往往急需调取数月前的对话原文却无从下手。本文介绍如何借助微信机器人开发技术，搭建一套自动化聊天记录备份系统，让每一条消息实时落库、可查可溯。

一、为什么需要自动备份，而不是手动导出

很多人的第一反应是"微信不是可以在电脑端导出聊天记录吗"。这个方案有几个硬伤：

操作频率限制：微信电脑端导出是手动触发行为，无法做到实时或按计划自动执行。一旦忘记操作，某段时间的记录就可能断档。

数据格式封闭：微信导出的是私有格式（.ftsmsg 或 HTML 页面），不便于程序化解析、全文检索和二次开发。图片、语音、文件等多媒体附件需要额外手动保存。

换设备即失效：聊天记录本地存储在手机 SQLite 数据库中，换机时如果迁移失败，旧设备的数据就再也回不来了。

不支持多账号聚合：企业通常运营多个微信号，分别手动导出效率极低，且难以做统一的关键词检索。

自动化备份机器人的核心思路是：通过个人微信API 实时监听消息事件，将每一条消息（包括文字、图片、语音、文件、撤回通知等）在产生的瞬间写入外部存储（数据库、对象存储等），完全绕开微信客户端本地存储的限制。

二、技术架构与核心原理

整套备份系统由三个层次构成：

微信账号（iPad 登录态）
      ↓
WechatApi 消息推送（WebHook / 长轮询）
      ↓
备份服务（消息解析 → 存储 → 通知）
      ↓
存储层（MySQL / MongoDB / OSS）

登录层：WechatApi 采用微信iPad协议维持账号在线状态。相比 PC 协议或 Web 协议，iPad 协议稳定性更高、被封风险更低，且支持接收所有消息类型（包括语音条、视频、小程序卡片等）。

消息层：WechatApi 在收到新消息时，会向你预先配置的 WebHook 地址主动推送一个 HTTP POST 请求，Body 是标准 JSON。你的备份服务只需要开一个 HTTP Server 监听这个地址即可。对于需要主动拉取的场景，也支持轮询接口按时间范围批量获取消息列表。

存储层：消息落库后，可按 sender_wxid + room_id + timestamp 建立复合索引，支撑后续的精确检索和批量导出。多媒体附件通过附件下载接口先拉到本地或上传至 OSS，再将 URL 存入数据库，避免消息记录中只有一个无法访问的临时链接。

通知层（可选）：备份异常（如磁盘满、接口超时）时，可通过微信机器人反向给管理员账号发送告警消息，形成自监控闭环。

三、环境准备与账号接入

在开始写代码之前，需要完成以下前置步骤：

注册 WechatApi 账号：访问 https://newmanager.wechatapi.net/dashboard/ 完成注册，获取 VideosApi-token（用于 API 鉴权）和 appId（代表一个具体的设备/账号实例）。

扫码登录微信：在控制台创建设备后，使用目标微信号扫码登录。登录成功后，该微信号的消息会实时推送给你的服务。

配置 WebHook 地址：在控制台的"消息推送"设置中填入你的服务公网地址（如 https://your-server.com/wechat/hook）。如果本地开发，可以先用 ngrok 做内网穿透。

准备存储：本文示例使用 MySQL，建议提前创建数据库和以下表结构。

sqlCREATE TABLE wechat_messages (
    id          BIGINT AUTO_INCREMENT PRIMARY KEY,
    app_id      VARCHAR(64)  NOT NULL COMMENT '设备ID',
    msg_id      VARCHAR(64)  NOT NULL UNIQUE COMMENT '消息唯一ID',
    room_id     VARCHAR(64)  NOT NULL COMMENT '会话ID(私聊=对方wxid,群聊=roomId)',
    sender_id   VARCHAR(64)  NOT NULL COMMENT '发送人wxid',
    msg_type    TINYINT      NOT NULL COMMENT '消息类型(1文字/3图片/34语音/…)',
    content     TEXT         COMMENT '文字内容或附件URL',
    raw_json    JSON         COMMENT '原始推送体',
    created_at  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    msg_time    DATETIME     NOT NULL COMMENT '消息实际发送时间',
    INDEX idx_room_time (room_id, msg_time),
    INDEX idx_sender_time (sender_id, msg_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

四、消息接收与解析实战（Python 示例）

下面用 Flask 实现一个最小可用的 WebHook 接收服务，演示消息解析和落库的完整流程。

pythonimport json
import logging
from datetime import datetime
from flask import Flask, request, jsonify
import pymysql

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

# ---- 数据库连接（生产环境请用连接池）----
DB_CFG = dict(host="127.0.0.1", port=3306, user="root",
              password="your_password", db="wechat_backup",
              charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor)

# 消息类型映射（微信协议定义）
MSG_TYPE_MAP = {
    1: "文字", 3: "图片", 34: "语音", 43: "视频",
    47: "表情", 49: "文件/链接/小程序", 10002: "撤回通知"
}

def save_message(app_id: str, msg: dict):
    """将单条消息写入 MySQL"""
    conn = pymysql.connect(**DB_CFG)
    try:
        with conn.cursor() as cur:
            sql = """
                INSERT IGNORE INTO wechat_messages
                    (app_id, msg_id, room_id, sender_id, msg_type,
                     content, raw_json, msg_time)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
            """
            cur.execute(sql, (
                app_id,
                msg.get("MsgId", ""),
                msg.get("FromUserName", ""),
                msg.get("SenderUserName", msg.get("FromUserName", "")),
                msg.get("MsgType", 0),
                msg.get("Content", "")[:65535],   # 截断超长文本
                json.dumps(msg, ensure_ascii=False),
                datetime.fromtimestamp(msg.get("CreateTime", 0))
            ))
        conn.commit()
        logging.info("saved msg_id=%s type=%s", msg.get("MsgId"), msg.get("MsgType"))
    finally:
        conn.close()

@app.route("/wechat/hook", methods=["POST"])
def wechat_hook():
    """WechatApi WebHook 入口"""
    body = request.get_json(force=True, silent=True) or {}
    app_id  = body.get("appId", "")
    messages = body.get("data", {}).get("list", [])  # 可能是批量推送

    if not messages and "MsgId" in body.get("data", {}):
        messages = [body["data"]]   # 单条推送兼容

    for msg in messages:
        try:
            save_message(app_id, msg)
        except Exception as e:
            logging.error("save failed: %s | msg=%s", e, msg.get("MsgId"))

    return jsonify({"ret": 200, "msg": "ok"})

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

代码说明几个关键点：

INSERT IGNORE 利用 msg_id 的唯一索引防止重复写入（网络重试时 WebHook 可能会重推）。
body.get("data", {}).get("list", []) 兼容批量推送和单条推送两种格式，实际以接口文档为准。
Content 字段对于图片/语音消息存的是媒体描述 XML，后续步骤需要解析 XML 再拉附件。

五、多媒体附件的下载与持久化

文字消息直接落库即可，但图片、语音、视频等附件在微信服务器上的存储时效有限（通常图片保留数天，语音更短），必须尽快下载到自己的存储。

以下示例演示如何调用 WechatApi 的附件下载接口，并上传至本地目录（生产环境建议换成 OSS）：

pythonimport requests
import os
import hashlib

API_BASE  = "https://api.wechatapi.net"   # 示意性地址，以实际文档为准
API_TOKEN = "your_videos_api_token"        # 控制台获取的鉴权 token
APP_ID    = "your_app_id"                  # 设备ID
SAVE_DIR  = "/data/wechat_attachments"

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

def download_media(msg_id: str, msg_type: int) -> str:
    """
    拉取附件并保存到本地，返回本地路径。
    msg_type: 3=图片 34=语音 43=视频
    """
    payload = {
        "appId": APP_ID,
        "msgId": msg_id,
        "type": msg_type
    }
    resp = requests.post(
        f"{API_BASE}/api/media/download",
        headers=HEADERS,
        json=payload,
        timeout=30
    )
    result = resp.json()
    # 标准返回体: {"ret": 200, "msg": "success", "data": {"base64": "...", "ext": "jpg"}}
    if result.get("ret") != 200:
        raise RuntimeError(f"附件下载失败: {result.get('msg')}")

    ext      = result["data"].get("ext", "bin")
    b64_data = result["data"].get("base64", "")
    import base64
    raw = base64.b64decode(b64_data)

    # 用内容哈希做文件名，避免重复存储
    sha  = hashlib.sha1(raw).hexdigest()
    path = os.path.join(SAVE_DIR, f"{sha}.{ext}")
    os.makedirs(SAVE_DIR, exist_ok=True)
    if not os.path.exists(path):
        with open(path, "wb") as f:
            f.write(raw)

    return path

返回体结构约定（与 WechatApi 所有接口一致）：

json{
  "ret": 200,
  "msg": "success",
  "data": {
    "base64": "iVBORw0KGgo...",
    "ext": "jpg",
    "size": 48291
  }
}

ret 非 200 时，msg 字段会描述具体错误原因，常见值包括 token无效、appId不存在、设备离线 等，建议在异常处理中记录完整返回体便于排查。

六、关键配置参数与接口速查表

参数 / 字段	类型	说明	示例值
`VideosApi-token`	请求头	API 鉴权 Token，在控制台"我的Token"中获取	`vat_xxxxxxxxxxxx`
`appId`	string	设备实例 ID，一个账号对应一个 appId	`wx_dev_abc123`
`MsgType`	int	消息类型，同微信协议定义	`1`（文字）
`MsgId`	string	消息唯一 ID，可用于幂等去重	`"8892341234"`
`FromUserName`	string	会话 ID（私聊为对方 wxid，群聊为群 ID）	`"wxid_xyz"`
`SenderUserName`	string	群聊中的实际发言人 wxid	`"wxid_sender"`
`CreateTime`	int	消息发送时间（Unix 时间戳，秒级）	`1718000000`
`Content`	string	文字内容；媒体消息为 XML 描述	`"你好"`
WebHook 推送间隔	—	控制台可配置：实时 / 5秒批量 / 10秒批量	建议实时
附件下载 base64 大小上限	—	单次返回建议不超过 10MB，超大文件需分片	—

消息类型编号速查：

MsgType	消息类型	备份重点
1	文字	直接存 Content
3	图片	下载原图 + 缩略图
34	语音	下载 AMR 文件，可选转码 MP3
43	视频	下载视频，注意存储空间
49	链接 / 文件 / 小程序	解析 Content XML 提取 url
10002	撤回通知	记录撤回事件，可标记原消息

七、定时巡检与补漏机制

WebHook 是事件驱动的，在极少数情况下（服务重启、网络抖动）可能丢失推送。为此，建议配合主动拉取接口做定时补漏：

bash# 用 curl 演示主动拉取最近 5 分钟内的消息列表
curl -X POST https://api.wechatapi.net/api/msg/list \
  -H "VideosApi-token: your_token" \
  -H "Content-Type: application/json" \
  -d '{
    "appId": "your_app_id",
    "startTime": 1718000000,
    "endTime":   1718000300,
    "pageSize":  100
  }'

返回结构同样遵循 {"ret":200,"msg":"...","data":{"list":[...],"total":N}} 范式。

补漏脚本建议每 5 分钟跑一次（用 crontab 或 APScheduler），对拉回的消息列表逐条执行 INSERT IGNORE，重复消息会被 msg_id 唯一索引自动过滤。

断点续传技巧：在 Redis 或数据库中维护一个 last_pull_time 键，每次补漏查询以上次成功时间为起点，避免每次都从头扫描。

八、注意事项与合规边界

稳定性：iPad 协议登录态受微信账号安全策略影响，长期在线的账号建议关闭"允许他人通过手机号搜索"等高风险设置，降低异常掉线概率。WechatApi 的微信二次开发文档中有详细的账号养护建议，上线前务必阅读。

存储规划：一个活跃微信号每天产生的文字消息约 1-5 万条，图片按均值 300KB/张计算，10 个群同时备份每月图片存储量约 30-100GB，需提前做磁盘规划。OSS 是比本地磁盘更稳妥的附件存储方案。

数据安全：聊天记录属于高度敏感的个人隐私数据，数据库必须启用加密存储（MySQL 透明加密 TDE 或字段级加密），传输层全程 HTTPS，访问日志完整保留。如果涉及企业用途，还需在员工知情的前提下部署，避免法律风险。

撤回消息的处理：MsgType=10002 是撤回通知，Content 中包含被撤回消息的 MsgId。你可以选择在数据库中将原消息标记 is_retracted=1 而不是物理删除，这样既保留了记录，又不会在前端检索时默认展示，比较符合合规审计的实际需求。

多账号隔离：如果同时备份多个微信号，务必在存储时携带 appId 字段做数据隔离，避免不同账号的消息混用 wxid（不同账号的好友 wxid 可能重复）。

小结

搭建微信聊天记录自动备份机器人的核心链路并不复杂：通过 WechatApi 基于 iPad 协议维持登录态 → WebHook 实时推送消息事件 → 解析入库 → 附件异步下载持久化 → 定时补漏兜底。整套方案无需 root 手机、无需微信官方开放接口，适用于个人记录保全、企业合规存档、客服质检等多种场景。

如果你的业务场景更进一步——不只是备份，还需要自动回复、关键词触发、群消息管理——可以在本套架构基础上扩展，WechatApi 提供的微信客服机器人和微信群管理机器人接口完全兼容本文的鉴权和消息格式，可以平滑叠加而无需重构。欢迎访问 https://wechatapi.net 了解全部接口能力，或前往开发文档查看最新接口规范。