简介这份资源面向熟悉Java开发、需要落地企业微信会话内容存档功能的技术人员围绕服务器配置、事件接收、公钥生成与配置、依赖引入、SDK使用以及消息拉取与解密等关键环节提供从零开始的完整实现思路。压缩包共20个文件约31KB以10个java源码为主体辅以3个md说明文档、2个yml配置、2个lst清单、1个class、1个xml及1个inscode工程文件覆盖代码、配置与说明三类内容便于直接对照项目结构理解实现方式。目前已有223人学习。资源重点给出可运行的代码示例并针对IP可信问题、依赖冲突处理、SO文件加载等常见难题提供解决思路帮助开发者减少踩坑、加快调试进度适合作为企业微信会话存档功能开发与排错的参考材料。1. 企业微信会话存档配置从开通到拉取消息的完整链路企业微信会话存档配置这件事很多团队第一次做都会卡在同一个地方后台明明开了开关代码拉消息却一直报错。我见过最典型的场景是某公司的后端同学花了两天排查网络和证书最后发现是可信 IP 没配——请求根本没到企业微信的服务器就被拦了。会话存档本质上是企业微信开放能力里偏「重」的一块它涉及密钥管理、SDK 调用、拉取游标维护、媒体文件解密四个环节任何一环没对齐整条链路就是黑匣子。它适合谁适合需要做合规审计、客服质检、销售会话分析的团队尤其是已经用企业微信做客户运营、想把聊天记录沉淀成结构化数据的公司。这篇笔记按「开通配置 → SDK 拉取 → 解密落库 → 排错」的顺序讲每一步都给可复现的命令和参数新手能照着跑通熟手能对照检查自己的边界条件。2. 开通与密钥配置三个参数决定你能不能拉到数据会话存档不是装个库就能用的东西它首先是一个需要企业管理员在后台手动开通的能力。开通之后你会拿到一组用于调用接口的凭证这组凭证的配置方式直接决定了后续代码能不能跑通。这一章把开通流程和密钥配置讲清楚因为后面所有代码都依赖这三个参数。2.1 后台开通会话存档与可信 IP 设置登录企业微信管理后台进入「管理工具 → 会话内容存档」这里会看到开通入口。开通时需要选择存档范围可以是全公司也可以指定部门或成员。我的建议是第一次先用一个小范围比如一个测试部门跑通确认链路没问题再扩大——因为存档一旦开启产生的数据量是按消息条数算的全量开启后再调试日志会被真实数据淹没。开通后最关键的一步是配置可信 IP。企业微信的接口调用有 IP 白名单机制你服务器出口 IP 不在白名单里所有请求都会返回60020之类的错误码。配置位置在会话存档的「API 配置」里把调用方的公网出口 IP 填进去。注意这里填的是出口 IP不是内网 IP如果你的服务部署在容器或 NAT 网关后面要确认实际出口地址。提示可信 IP 支持配置多个用换行分隔。调试阶段可以临时把你的开发机出口 IP 加进去但上线前记得清理避免留下不必要的入口。配置完成后后台会生成Secret用于获取 access_token和RSA 公钥用于加密对称密钥。这两个东西加上企业 IDCorpID就是后续代码里的三个核心参数。2.2 三个核心参数的含义与获取方式把这三个参数理解清楚后面看代码就不会懵参数来源作用注意事项CorpID企业微信后台「我的企业」标识企业身份固定值不随成员变化Secret会话存档「API 配置」生成换取 access_token泄露等于别人能拉你的存档务必放密钥管理服务RSA 公私钥后台生成或自行上传公钥加密/解密对称密钥私钥自己保管公钥配置到后台获取 access_token 的请求长这样用 curl 就能验证# 用 CorpID 和 Secret 换取 access_token # corpid 和 corpsecret 替换成你自己的值 curl -s https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpidYOUR_CORPIDcorpsecretYOUR_SECRET返回结果里access_token有效期是7200 秒需要缓存并在过期前刷新。这里有个血泪经验不要每次请求都去换 token企业微信对获取 token 的频率有限制频繁调用会触发限流表现为后续接口间歇性失败。常见做法是用 Redis 缓存设置 7000 秒过期加一个分布式锁防止并发刷新。2.3 配置对称密钥RSA 加密那一步到底在做什么会话存档的媒体文件图片、语音、文件不是明文传输的企业微信为每个会话生成一个随机的对称密钥然后用你配置的RSA 公钥加密后传给你。你要做的就是用私钥解密拿到对称密钥再用它解密媒体内容。这一步的配置动作是在后台「API 配置」里填入你的 RSA 公钥。公钥格式要求是PKCS#1或PKCS#8的 PEM 格式很多同学在这里翻车——用工具生成的默认是 PKCS#8但后台可能只认某一种报错信息又不明确。我的做法是统一用 OpenSSL 生成并显式指定格式# 生成 2048 位 RSA 私钥PKCS#1 格式 openssl genrsa -out private_key.pem 2048 # 从私钥导出公钥 openssl rsa -in private_key.pem -pubout -out public_key.pem # 查看公钥内容确认是 PEM 格式 cat public_key.pem生成后把public_key.pem的内容完整粘贴到后台配置框。私钥private_key.pem保存在服务端后面解密对称密钥要用。注意私钥不要提交到代码仓库用环境变量或配置中心注入。3. SDK 拉取消息游标、批量与媒体文件处理配置通了接下来是真正拉数据。会话存档的拉取不是「查询历史消息」那种模式而是基于游标的增量拉取你告诉它从哪个位置开始拉它返回一批消息和一个新的游标下次用新游标继续。这个设计决定了你的代码必须自己维护游标状态否则要么重复拉取要么漏消息。3.1 拉取接口的调用方式与游标维护企业微信提供了 SDKC/Java/Python 等也支持直接调 HTTP 接口。SDK 的好处是帮你处理了加解密和协议细节坏处是部署依赖多。我一般先用 HTTP 接口验证链路确认没问题再换 SDK。拉取的核心接口是getchatdata关键参数seq起始游标第一次传 0limit单次拉取条数最大 1000proxy和passwdSDK 模式下用于解密HTTP 模式下传空import requests import json def pull_messages(access_token, seq0, limit100): 从指定游标拉取会话存档消息 url https://qyapi.weixin.qq.com/cgi-bin/msgaudit/getchatdata params {access_token: access_token} body { seq: seq, limit: limit, # HTTP 模式下 proxy 和 passwd 留空 proxy: , passwd: } resp requests.post(url, paramsparams, jsonbody) data resp.json() if data.get(errcode) ! 0: raise RuntimeError(f拉取失败: {data}) # 返回消息列表和下一个游标 return data.get(chatdata, []), data.get(next_seq, seq)逻辑说明seq是游标第一次传 0之后每次用返回的next_seq。limit控制单次批量建议设 500 到 1000太小会导致请求次数多、触发限流太大则单次响应体过大、容易超时。返回的chatdata里每条消息包含发送者、接收者、消息类型、时间戳以及加密后的内容。游标维护是这一环最容易出问题的地方。我的做法是每拉一批就持久化一次游标用数据库或 Redis 存key 用企业 ID 加模块名。这样即使服务重启也能从上次位置继续。千万不要把游标只放在内存里进程一挂就全丢了。3.2 消息解密从密文到可读内容的完整流程拉回来的chatdata里encrypt_random_key和encrypt_chat_msg两个字段是加密的。解密分两步第一步用 RSA 私钥解密encrypt_random_key得到对称密钥。注意这个字段是Base64 编码的先解码再解密。第二步用对称密钥解密encrypt_chat_msg得到明文消息体。企业微信 SDK 里封装了这个过程如果走 HTTP 接口需要自己实现。import base64 from Crypto.PublicKey import RSA from Crypto.Cipher import PKCS1_v1_5 def decrypt_random_key(encrypt_random_key, private_key_pem): 用 RSA 私钥解密对称密钥 # encrypt_random_key 是 Base64 编码的密文 ciphertext base64.b64decode(encrypt_random_key) private_key RSA.import_key(private_key_pem) cipher PKCS1_v1_5.new(private_key) # 解密后得到对称密钥用于后续消息解密 # 注意这里需要配合企业微信提供的解密库处理填充 return cipher.decrypt(ciphertext, None)参数说明encrypt_random_key来自拉取接口返回private_key_pem是你之前生成的私钥。解密出的对称密钥是一个字节串直接传给消息解密函数。消息解密后是 JSON 格式包含msgtype、text、image等字段按类型解析即可。注意RSA 解密对填充格式敏感如果报解密失败先检查私钥格式是否和后台配置的公钥匹配。常见错误是用 PKCS#8 私钥去解 PKCS#1 公钥加密的数据虽然都是 RSA但填充方式不同会直接失败。3.3 媒体文件拉取与本地落盘文本消息解密后直接就是内容但图片、语音、文件这类媒体消息解密后拿到的是一个sdkfileid需要再调一次接口下载。def download_media(access_token, sdkfileid, save_path): 根据 sdkfileid 下载媒体文件 url https://qyapi.weixin.qq.com/cgi-bin/msgaudit/getmediafile params {access_token: access_token} body {sdkfileid: sdkfileid} resp requests.post(url, paramsparams, jsonbody) if resp.headers.get(Content-Type) application/json: # 返回 JSON 说明出错了 raise RuntimeError(f下载失败: {resp.json()}) # 二进制内容直接写文件 with open(save_path, wb) as f: f.write(resp.content)逻辑说明这个接口返回的是二进制流不是 JSON。判断方式看Content-Type如果是application/json说明返回了错误码。sdkfileid有有效期一般是几天所以拉取到媒体消息后要尽快下载不要攒着批量处理。落盘路径建议按日期和会话 ID 分目录避免单目录文件过多导致文件系统性能下降。4. 避坑与排查那些让你怀疑人生的错误码会话存档的报错信息普遍不友好很多错误码只给一个数字不告诉你哪里错了。这一章整理几个我实际踩过的坑按「现象 → 原因 → 解决」写希望能帮你省下几个小时的排查时间。4.1 坑一60020 错误——可信 IP 没生效现象所有接口都返回errcode: 60020提示 IP 不在白名单。原因可信 IP 配置后不是立即生效有几分钟延迟或者你填的是内网 IP实际出口是另一个地址。解决先用curl ifconfig.me确认服务器真实出口 IP再和后台配置比对。如果用了容器或网关确认是网关出口 IP。配置后等 5 分钟再试别急着改代码。4.2 坑二拉取返回空但明明有消息现象getchatdata返回errcode: 0但chatdata是空数组。原因游标seq传错了。如果你之前已经拉到某个位置再用 0 去拉可能因为数据保留策略拿不到旧数据或者存档范围没包含目标成员。解决检查游标是否从上次的next_seq继续。确认存档范围配置里包含了你要拉取的成员。企业微信的存档数据有保留期限超期的拉不到。4.3 坑三媒体文件下载返回 JSON 错误现象调getmediafile返回的不是二进制而是{errcode: xxx}。原因sdkfileid过期了或者 access_token 权限不足。解决sdkfileid有效期短收到媒体消息后立即下载。access_token 要用会话存档对应的 Secret 换取不要用其他应用的 token。4.4 坑四解密报填充错误现象RSA 解密时抛ValueError: Incorrect decryption或类似填充错误。原因私钥和后台配置的公钥不匹配或者私钥格式不对。解决重新生成一对密钥公钥重新配置到后台私钥更新到服务端。确认生成命令用的是同一套参数。如果之前配置过旧公钥先删除再添加新的。4.5 坑五高频拉取触发限流现象跑一段时间后接口开始间歇性失败过一会又恢复。原因请求频率超过企业微信限制。会话存档接口有调用频次上限具体数值随企业规模变化。解决拉取循环里加 sleep比如每批之间停 200 毫秒。access_token 用缓存不要每次重新获取。如果数据量大考虑分时段拉取避开业务高峰。5. 进阶把拉取做成可持续运行的管道跑通单次拉取只是开始真正要落地的是7×24 小时稳定运行的管道。这一章讲几个让管道更可靠的具体技巧都是我在实际运维里攒下来的。5.1 用状态机管理拉取生命周期把拉取逻辑拆成「获取 token → 拉取批次 → 解密 → 落库 → 更新游标」几个状态每个状态失败都有明确的恢复路径。比如 token 过期就重新获取拉取失败就重试当前游标解密失败就记录原始密文到死信队列后续人工处理。这样任何一环出问题都不会导致整个管道卡死。import time import redis r redis.Redis() def run_pipeline(): 持续拉取的主循环 while True: try: token get_cached_token() # 从缓存获取过期自动刷新 seq int(r.get(chat_archive_seq) or 0) messages, next_seq pull_messages(token, seq) for msg in messages: try: plain decrypt_message(msg) save_to_db(plain) except Exception as e: # 单条失败不影响整批记录后继续 r.lpush(chat_archive_dlq, json.dumps(msg)) print(f单条处理失败: {e}) # 整批成功后更新游标 r.set(chat_archive_seq, next_seq) # 控制频率避免限流 time.sleep(0.2) except Exception as e: print(f批次失败10 秒后重试: {e}) time.sleep(10)逻辑说明游标存在 Redis 里整批处理成功才更新保证至少一次语义。单条失败进死信队列不阻塞后续消息。批次失败等 10 秒重试避免雪崩。time.sleep(0.2)是频率控制根据实际限流情况调整。5.2 监控指标怎么知道管道还活着管道跑起来之后你需要知道它是不是在正常工作。我一般监控四个指标指标含义异常阈值拉取延迟当前时间与最新消息时间差超过 10 分钟告警死信队列长度处理失败的消息数持续增长告警游标推进速度每分钟游标增量长时间为 0 告警token 刷新失败次数获取 token 失败计数连续 3 次失败告警这些指标用 Prometheus 或简单的定时脚本都能采集。关键是拉取延迟它直接反映管道是否跟得上消息产生速度。如果延迟持续增大说明拉取速度不够需要调大limit或增加并发。5.3 数据落库的表结构设计消息落库建议分两张表一张存消息主体一张存媒体文件索引。消息主体表按时间分区字段包括消息 ID、发送者、接收者、消息类型、内容、时间戳。媒体文件索引表存 sdkfileid、本地路径、文件类型、关联的消息 ID。-- 消息主体表按月份分区 CREATE TABLE chat_message ( id BIGINT AUTO_INCREMENT, msg_id VARCHAR(64) NOT NULL, sender VARCHAR(64), receiver VARCHAR(64), msg_type VARCHAR(32), content TEXT, msg_time DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id, msg_time), KEY idx_msg_time (msg_time), KEY idx_sender (sender) ) PARTITION BY RANGE (TO_DAYS(msg_time)) ( PARTITION p202401 VALUES LESS THAN (TO_DAYS(2024-02-01)), PARTITION p202402 VALUES LESS THAN (TO_DAYS(2024-03-01)) );参数说明msg_id加唯一索引可以防重复插入。按时间分区方便清理旧数据和按时间范围查询。content用 TEXT 存明文内容如果合规要求加密存储在这一层做字段级加密。5.4 一个容易忽略的细节时区企业微信返回的时间戳是Unix 时间戳单位秒。落库时如果用FROM_UNIXTIME转换注意数据库时区设置。我遇到过服务器时区是 UTC、数据库时区是东八区导致消息时间差了 8 小时排查了半天。统一做法是存储时用 UTC展示时在前端转换。或者干脆存时间戳整数查询时再格式化。这个管道我维护了一年多最大的教训是不要相信任何一次成功。每次企业微信接口调整、每次服务器迁移、每次证书更新都可能让管道悄悄挂掉。所以监控和告警不是可选项是必选项。希望帮到你。本文还有配套的精品资源点击获取
