企微会话存档开发实战：消息拉取、解密与群聊数据落库

1. 为什么要把群聊消息“存档”出来

做企业服务或者内部管理的同学，应该都遇到过这类需求：销售团队和客户的沟通记录需要定期抽检，客服群里的服务质量要复盘，项目群里讨论的关键信息散落在聊天记录里没法做统计。这个时候，“群内信息提取”就成了刚需。而企微会话存档，就是官方提供的一条合规路径，它不是让机器人模拟登录去捞聊天记录，而是通过管理员授权、接口拉取的方式，把整个组织内符合条件的会话数据拿回来自己解析和存储。

我第一次接触这个功能，是在做某客户服务系统的对接时。当时客户提了个很具体的要求：把某个核心客户群里的全部消息同步到他们的内部系统，跑质检规则，还要按月出服务报告。我第一反应是“这不就是个消息读取吗”，结果一查文档才发现，企微的会话存档并不像看起来那么简单——涉及企业侧配置、接口鉴权、消息加解密、增量拉取、数据落库，每一步都有自己的坑。

这篇文章把整个链路拆开讲一遍。适用对象是：准备自己写代码拉取存档数据的开发者，或者负责为企业内部搭建会话记录分析系统的实施人员。读完你应该能搞清楚这套机制怎么运转、代码怎么写、数据落库大概怎么设计，以及最容易在哪个环节翻车。

2. 前期准备：开通、配置、拿凭证

2.1 开通会话存档并设置可见范围

会话存档不是默认开通的能力。按企业管理的操作路径，通常需要管理员登录管理后台，在“管理工具”或类似模块里找到会话存档的入口，然后阅读协议、确认开通。开通之后有一个非常关键的配置：设置存档范围。范围决定哪些成员的聊天消息会被存档，这个范围必须是部门、成员或标签的集合。

我建议你第一轮配置时只选一个较小的范围——比如某个测试部门或者三五个人。原因是等会儿调试接口时，如果范围太大，消息量大、日志杂，出了问题很难定位；小范围内跑通了，再逐步扩大归档范围。另外，企微的存档机制对于“合规存档”是有明确边界的：只有被纳入范围的成员，其会话消息才会被推送。群聊里如果有人不在范围内，他和别人的消息不一定会完整归档，这个在最终方案设计时要跟业务方说清楚，避免验收时扯皮。

2.2 从管理后台拿到关键凭证

开通之后，最关键的一步是收集凭证。做会话存档开发，你至少要拿到这几样东西：

• 企业ID（corpid）：企业唯一标识，所有接口都会用到。
• 会话存档Secret（secret）：调用专用接口的密钥，跟普通应用的secret不是一个概念，注意区分。
• 公钥：用于加密存档内容中的AES密钥，企微侧使用公钥加密，你这边要用对应私钥解开。
• 私钥：由企业自行生成，然后上传公钥到管理后台。有些平台的流程是自助生成公私钥对，把公钥内容填进后台，私钥由企业自己妥善保存。

这块有个细节值得多说几句。很多第一次搞的同学会误以为加密密钥是自己的普通应用密钥，尤其看到代码里写了一堆key_name、version之类的字段，容易懵。实际上，企微会话存档的数据保护设计是“双层”的：传输时用RSA保护一把随机的AES密钥，再用这个AES密钥去加密真正的消息体。所以后台配置的那把公钥，是用来加密随机AES密钥的，不是用来直接加密消息内容的。理解这层关系，后面看协议才不会云里雾里。

2.3 回调配置与IP白名单

写一个接收存档数据的服务前，得先准备一个公网可访问的回调地址，并把它配置到相关的接收事件或对接参数中。这个怎么验证、怎么配置，不同版本的管理后台略有差异，但大体逻辑都是：填一个URL，企微后台会往这个地址发一条验证消息，你的服务收到后需要按既定规则返回响应，验证通过才算生效。

另外要留意IP白名单。调用接口的服务器出口IP必须提前加白，特别是你用了云函数、容器等弹性IP时，IP漂移会导致接口突然报错。我把这个写进笔记里，因为排查过好几次“明明密钥没问题怎么401/6000”的报错，最后发现就是出口IP变了。

3. 核心原理拆解：加解密链路到底是怎么走的

3.1 消息不是明文给到你

很多人一开始会下意识以为，接口拉回来的一定是现成的JSON明文。实际上拉回来的数据是密文块，需要先解密才能得到我们熟悉的聊天内容字段。这样设计的理由很直接：聊天记录涉及具体人员的沟通内容，属于高度敏感信息，平台不可能在传输链路上泄露明文，所以必须在数据层做可靠加密。

你拉取一条原始消息时，得到的是一串二进制封装，里面至少包含：

• 消息体头部：可能是版本号、类型、长度之类的元数据；
• 加密后的AES密钥：这是用企业上传的那把公钥做RSA加密得到的；
• 真正用AES加密过的消息内容；
• 消息相关的接收者标识、校验信息等。

作为开发者，你不需要深究每一个字节的排列是否符合某些内部规范，但你必须知道这个结构的存在——因为几乎所有官方SDK或者开源实现，最先做的就是“拆包”：拆出长度、拆出AES密钥密文、拆出密文内容。这个拆包过程如果整体错位，解密出来的就是乱码。

3.2 解密时的两把钥匙怎么用

第一步，用企业私钥做RSA解密，得到一个AES密钥原文。这个AES密钥每一条消息都可能不同，是随机生成的一次性密钥。第二步，拿着这个AES密钥，对消息体中的密文做对称解密，得到原始消息明文。

你可以把这个过程理解成寄快递：RSA是“保险柜钥匙”，AES是“快递箱密码”。平台先把快递箱密码放进保险柜，然后把保险柜寄给你；你用自己的私钥打开保险柜拿到密码，再用密码打开快递箱拿到内容。双层保护的目的是兼顾安全性和效率——RSA安全但慢，不适合加密大内容，所以只用来保护那把一次性密码；AES快，适合加密真正的消息正文。

3.3 为什么需要自己造轮子

官方SDK确实提供了解密封装，但我强烈建议你自己写一个最小可用的解密单元再对接SDK，至少把解密流程的每个环节打印出来逐步验证。理由很简单：一旦SDK返回decrypt failed，如果不懂底层流程，你完全不知道是RSA失败还是AES失败，还是消息结构被截断了。我自己就吃过这个亏，后来拆开每一步做日志打点，才知道是私钥格式不对——PKCS#1和PKCS#8混用，半天没排查出来。

4. 核心代码实现：从拉取到落库

4.1 获取access_token的基础流程

会话存档的接口跟企微其他开放接口一样，都需要access_token。获取token的常见方式是调用企业凭证接口，传corpid和corpsecret，同一个secret调用会有一个有效期，一般是7200秒。这里的关键点是：token的获取不要“每次都现拿”，否则一旦消息量大、拉取频率高，会被限流。

我的做法是在本地做一个token缓存，以corpsecret为key，把access_token和过期时间存到Redis或者内存里，提前5分钟过期重建。这样既降低了凭证接口的调用量，也避免由于瞬时过期导致的消息拉取中断。

代码逻辑比较简单，核心就是缓存判断：

def get_access_token(corp_id, secret, http_client=None): cache_key = f"wework_token:{corp_id}:{secret}" token = cache_get(cache_key) if token: return token url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken" params = {"corpid": corp_id, "corpsecret": secret} resp = http_client.get(url, params=params) data = resp.json() if data.get("errcode") == 0: cache_set(cache_key, data["access_token"], data["expires_in"] - 300) return data["access_token"] else: raise RuntimeError(f"get token failed: {data}")

需要提醒的是，企微的接口返回虽然普遍是HTTP 200，但业务状态都在JSON的errcode字段里。所以判断成功与否绝不能只看resp.status_code，一定要检查errcode。

4.2 拉取消息的接口调用要点

拉取存续消息的接口，核心参数一般包括：token、limit、type、start_time和end_time或seq游标。不同版本接口不同，但大体区分两套：

• 按时间范围拉取：适合离线补数据，但要注意单次拉取的消息量有上限，且时间范围太大会被拒绝。
• 按游标增量拉取：适合实时同步，通过保存最后一次拉到的位置，不断循环获取新消息。

我在实际项目里优先用游标方式，因为做实时同步更稳定，不需要关心时间窗口。流程就是：

1. 初始化游标为0；
2. 调用拉取接口，带回一批数据和新的游标；
3. 把当前游标存到数据库或Redis；
4. 循环拉取直到返回数据为空；
5. 间隔几秒再拉下一轮。

游标存哪也要想清楚。实测中，如果遇到服务重启，游标丢失会导致消息拉重或漏拉。我用过Redis的持久化存储，也见过有人直接存在MySQL表里，都可以，前提是“先落游标，再处理消息”，这样才能保证消息不丢。

4.3 解密流程的完整代码示意

下面这段代码是我整理过的一个简化版解密流程，去掉了一些繁琐的日志和异常分支，但基本能跑通整个链路。加密库使用的是常见的pycryptodome。

import base64 import json import os from Crypto.Cipher import AES, PKCS1_v1_5 from Crypto.PublicKey import RSA def parse_raw_message(raw_bytes, private_key_pem): # 1. 解析消息头：通常前面若干个字节表示消息体长度 msg_len = int.from_bytes(raw_bytes[:4], byteorder="big", signed=False) # 2. 取出加密后的AES密钥区（长度需要对照协议，常见是256字节） encrypted_aes_key = raw_bytes[4:260] # 3. 取出后续的密文消息（长度由msg_len决定） cipher_text = raw_bytes[260:260 + msg_len] # 4. 用私钥解密AES密钥 rsa_key = RSA.importKey(private_key_pem) cipher_rsa = PKCS1_v1_5.new(rsa_key) aes_key = cipher_rsa.decrypt(encrypted_aes_key, None) # 5. 用AES密钥解密消息体 nonce_size = 16 # 随机串长度，具体以实际解析为准 iv_and_payload = cipher_text iv = iv_and_payload[:nonce_size] encrypted = iv_and_payload[nonce_size:] aes = AES.new(aes_key, AES.MODE_CBC, iv) plain_padded = aes.decrypt(encrypted) plain_text = strip_padding(plain_padded) # 6. 解出业务JSON msg_json = json.loads(plain_text.decode("utf-8")) return msg_json

这里我要特别说明一点：不同加密长度、随机串长度、字段排列顺序在不同的SDK版本里可能不完全一致，上面代码中的数字基于我接触过的常见实现，不是所有平台版本都通用。正确做法是先读官方SDK文档给出的封装结构说明，再对照调整偏移量。写demo的时候，可以先打印raw_bytes的前几十个字节，对比文档里的字段说明，确认没理解偏。

4.4 消息落库的结构设计

解密出来的消息是一个JSON，里面通常包含序号、发送人、接收人、消息类型、内容、时间、群聊标识等字段。落库表结构建议考虑这几个字段：

CREATE TABLE msg_archive ( id BIGINT PRIMARY KEY AUTO_INCREMENT, msg_seq BIGINT, msg_type VARCHAR(20), room_id VARCHAR(64), from_user VARCHAR(64), to_user VARCHAR(64), content TEXT, msg_time DATETIME, raw_data LONGTEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_msg_seq (room_id, msg_seq) );

msg_seq是消息的唯一序号，同一会话内应该递增。加联合唯一索引的目的，是做幂等——万一游标写乱了，重复拉取也不会产生重复数据。raw_data字段保存解密后的完整原始JSON，方便以后回溯排查。

群聊消息群里的room_id通常对应一个群聊会话标识，单聊记录可能没有room_id或者room_id为空，这个要按实际返回判断。具体到项目实现，建议你可以先拉一批样本消息，把字段结构整理出来再定表结构，比直接照抄网上的建表语句靠谱得多。

5. 群聊消息的特殊处理

5.1 怎么识别一条消息是不是群聊

拉到的消息里，重点看room_id或类似的会话标识字段。有群聊标识的，说明这条消息属于群会话；没有的话则是单聊或外部联系人会话。我的测试经验是：先构造一个“只有自己和一个同事”的群，发几条消息，拉出来看字段；然后再建一个真正多人的群，发消息，对比差异，很快就能搞懂字段规律。

另外，群聊里有一个非常容易踩坑的地方：消息里的接收人字段。群聊场景下接收人可能是一个群标识，而不是具体某个成员ID，这时如果你拿着接收人字段去匹配组织架构，必然查不到人。正确做法是把群标识存成room_id，单独建一张群成员映射表，后续渲染会话列表时再做关联。

5.2 群成员与roomid的关联办法

如果你想做“某个群里谁发了什么”的统计，只靠拉消息还不够。你需要一张群成员关系表，把room_id和成员ID对应起来。这个关系怎么拿？通常有两种途径：

• 在收到每条消息解析时，如果消息体里携带了发送者和群标识，可以做增量积累；
• 某些接口版本支持直接获取群成员，但需要额外权限和不同接口逻辑。

我在实际项目中遇到过“群里有人退出后再进新群”的情况，同一人可能在同一房间标识下有多个进进出出的时间段。所以关系表至少要记录加入和离开时间，否则统计出来的“群人数”会虚高。这个问题初期很难发现，等后期做数据报表时才会暴露。

5.3 各类消息类型的差异

群聊里的消息类型远比想象中多。文字消息只是最基础的，还有图片、语音、视频、文件、链接、小程序、撤回通知等。不同类型的消息，content字段的路径完全不同。

文字消息一般是直接JSON里的content字段；图片和文件类消息，往往只返回一个file_id或media_id，你得再去调用文件下载接口才能拿到实际内容；撤回消息则是一条特殊的事件消息，记录的是被撤回消息的msgid，而不是内容本身。

我的建议是，第一版先把文字消息跑通，同时把其他类型的消息统一落库为“非文本消息”，只存元数据。然后根据业务需要，逐个类型去接下载和转存逻辑。不要一上来就想把所有消息类型全都处理完，那样只会让自己陷入无休止的适配工作中。

5.4 多媒体消息能不能拿到

多媒体消息能否拿到，取决于你的存档范围配置和接口权限。有些类型需要额外申请权限，或者受限于企业的合规配置。这里没有通用的固定答案，只能建议你把权限配置逐项确认清楚，再开发对应模块，避免辛辛苦苦写完代码，结果线上拉不到数据。

我这里就不列那些具体的类型清单了，因为不同版本差异很大，而且可能随平台策略调整。真实做法是：先用线上数据打印出所有出现的msg_type，整理成一张清单，再逐个确认哪些需要特殊处理。这样既稳又不会做无用功。

6. 常见问题与排查实录

6.1 问题速查表

这个表是我自己整理的心得，不一定适用所有版本的接口，但排查思路是通用的：遇到异常先分“是鉴权问题”“是协议问题”“是数据问题”，再逐层缩小范围。

6.2 解密失败到底怎么看

解密失败是最劝退新手的环节。很多同学看到decrypt fail就没了思路。我建议按以下顺序排查：

1. 先确认私钥是否跟后台公钥配对。可以拿一段已知明文的测试块，先用公钥加密再用私钥解密，能解回来，说明密钥对没问题。
2. 再确认消息拆包偏移量是否正确。在解密代码中把raw_bytes的hex形式打印出来，跟文档里各段长度算一遍，看拼接的位置对不对。
3. 还要确认AES的模式和对齐方式。比如加密模式、填充方式、IV长度都不一样，逐项核对。

有一次我排查半天，最后发现是环境里缺少加密库的底层依赖，导致AES.new()走了伪随机数生成，解密结果是“看似随机”的乱码。所以出问题时不要总觉得是协议理解错，也要看依赖库是否被正确安装。

6.3 拉取消息遇到稀疏空窗怎么办

高峰时段拉到大量消息，凌晨可能一条都没有。这时要注意拉取循环千万别因为“没消息”就退出，否则新消息来了就再也拉不到了。正确做法是设置一个空闲间隔参数，比如连续N次拉到空数据后，sleep一下再继续拉。

这里还有一个细节：有些接口返回“空数据”时，游标可能保持不变，也可能阶段性跳跃。如果游标保持不变，你其实没法只通过游标判断“当前是否有新数据”，必须靠不断轮询的机制来兜底。

6.4 数据量大了之后怎么优化

当消息量积累到几百万条后，再逐条循环插入数据库就会非常慢。我后来做了一些调整：

• 批量插入：把解密后的消息攒够100条或者200条，用一次SQL批量写入，性能提升非常明显；
• 索引精简：不要为了图方便把所有字段都建索引，只给查询频繁的字段建索引，比如msg_seq和room_id；
• 归档冷热分离：超过一定时间的历史消息，从主表搬到归档表，保持主查询高效。

实时拉取和离线补拉的逻辑可以拆成两个服务进程。一个持续跑游标增量，一个在业务低谷期用时间范围补拉历史数据，避免互相影响。

7. 我自己踩过的一些坑

这里分享几个最容易被忽视的经验，也算是对后面接手的人的一点提醒。

第一，私钥安全要当成最高优先级。会话存档的解密私钥一旦泄露，意味着所有聊天内容都有可能被第三方读取，这个风险是致命的。我的处理办法是：私钥不进代码仓库，放在专门的配置中心或密钥管理服务里，服务启动时动态读取，权限最小化。

第二，文档版本的坑。企微会话存档相关接口的文档有过多次调整，包括参数名、返回结构、加密细节。网上的技术文章很多，但很可能写得是老版本。我建议以官方最新文档为准，我刚看过的那些示例代码只做参考，不能直接拿生产环境用。

第三，日志要分级打点。我把解密服务的日志分成了三类：调试日志打印拆包前后的hex和长度，用来定位协议问题；业务日志记录拉取轮次、游标位置、消息条数，用来观察健康状况；错误日志只记录异常堆栈和相关上下文。这三类日志的区分让我在线上排查问题的时候，效率比原来高了不少。

最后说一下设计目标。整个会话存档提取方案，本质上不是“爬一个接口”那么简单，而是一个需要稳定持续运行的数据管道。拉取的稳定性、解密的正确性、存储的幂等性，每一项都需要提前设计。但反过来，你只要把token、游标、解密、落库这四件事做扎实了，这套系统是可以跑得很稳的。我自己在线上跑了大半年，中途除了例行维护，几乎没有再动过核心逻辑。

如果你正在做类似需求，我建议不要畏惧加解密的复杂度，先搭一个最小链路，用几天真实消息把全流程串起来，再慢慢完善。方向对了，后面的细节都好补。