简介这是一套面向企业微信二次开发者的会话存档源代码聚焦企微聊天记录合规留存与内容风控场景适合具备Java基础、需要落地会话审计功能的开发者参考。代码遵循官方解析处理流程采用多线程同步机制提升拉取效率默认实时记录seq队列值以支持增量运行并可动态同步指定时间范围的数据。资源包共42个文件约9.87MB以21个Java源码为核心辅以9个jar依赖、4个dll与1个so本地库、yml与properties配置、xml及md说明文档结构完整、注释清晰可开箱即用。功能上覆盖cos文件上传、es数据存储与敏感词过滤形成从数据拉取、解析、存储到内容筛查的闭环。目前已有2104人学习下载读者可据此快速理解企微会话存档的接入方式与工程组织掌握增量同步、文件转存与敏感词拦截的实现思路减少从零搭建的成本。1. 企业微信会话存档源代码从拉取到落库一条能跑通的最小链路企业微信会话存档这件事真正卡住大多数团队的从来不是「要不要做」而是拿到源代码之后发现跑不起来。官方给的是 SDK 和加解密库不是一套开箱即用的服务网上流传的所谓「企业微信会话存档源代码」多半是某个项目的片段缺了拉取调度、媒体分片下载、密钥管理和落库这几块直接编译能过一接真实企业就翻车。这篇讲的就是把这条链路补全从 corpid、secret、私钥三件套开始到 media_id 拉取、sdk 解密、消息分表入库最后能对着一个会话窗口验证「这条消息确实存下来了」。适合正在做合规归档、客服质检、销售会话分析的后端和运维也适合想评估这套方案到底值不值得投入的技术负责人。下面所有代码都是可替换参数直接跑的最小实现不依赖任何特定云厂商。2. 会话存档的拉取模型为什么不能只写一个 while 循环2.1 sdk 拉取的本质是「游标 分片」不是长连接企业微信会话存档的官方 SDK 提供的核心接口是GetChatData它的语义是你给它一个seq起始序号和一个limit单次条数它返回一批加密的消息体和下一个seq。很多人第一反应是写个while(true)不停拉结果要么被限频要么 seq 处理错导致消息重复或丢失。正确的模型是把 seq 当成一个持久化的游标每次拉取成功后先落库再推进游标任何一步失败都不能推进。这个顺序是血泪经验——先推进游标再落库一旦入库失败那批消息就永久丢了官方不会给你补。拉取还有两个隐藏约束。第一limit官方建议不超过 1000实际生产里我一般设 200 到 500因为单批太大时解密和入库的耗时会让整个循环的节奏失控反而更容易触发限频。第二seq 是全局递增的但不同会话的消息会交错返回所以你不能假设「这一批都是同一个会话的」入库时必须按roomid或from做分流。2.2 用 Python 跑通第一次拉取先装依赖官方 SDK 是 C 编译的动态库Python 侧一般用 ctypes 封装或者用社区维护的绑定。这里给一个 ctypes 直接调用的最小例子参数名和官方文档一致。import ctypes import json # 加载官方 sdk 动态库路径按实际部署改 sdk ctypes.CDLL(./libWeWorkFinanceSdk.so) # 初始化 sdk返回一个 handle sdk.NewSdk.restype ctypes.c_void_p handle sdk.NewSdk() # 初始化企业信息corpid secret # 注意 secret 是会话存档专用的不是应用 secret init_ret sdk.Init(handle, byour_corpid, byour_chatdata_secret) if init_ret ! 0: raise RuntimeError(finit failed: {init_ret}) # 准备拉取参数 seq 0 # 起始游标首次从 0 开始 limit 200 # 单批条数生产建议 200-500 proxy b # 内网环境可留空 passwd b # 私钥密码没设就留空 timeout 30 # 秒 # 调用 GetChatData buf ctypes.create_string_buffer(1024 * 1024 * 4) # 4MB 缓冲 ret sdk.GetChatData(handle, seq, limit, proxy, passwd, timeout, buf, len(buf)) if ret ! 0: raise RuntimeError(fgetchatdata failed: {ret}) data json.loads(buf.value.decode(utf-8)) print(next_seq:, data[seq]) print(msg_count:, len(data[chatdata]))这段代码的逻辑说明NewSdk和Init是必须的前置Init返回非 0 说明 corpid 或 secret 不对或者这个 secret 没有会话存档权限。GetChatData的返回值里seq是下一批的起始游标chatdata是加密消息数组。参数上limit不要贪大timeout设 30 秒足够内网代理留空即可。跑通这一步你就能看到加密的encrypt_random_key和encrypt_chat_msg接下来才是解密。3. 解密与媒体下载私钥、随机密钥、sdk 三层怎么串3.1 解密链路的三层结构会话存档的消息解密不是一步是三步。第一步用你生成密钥对时拿到的私钥对encrypt_random_key做 RSA 解密得到一个中间密钥。第二步用这个中间密钥对encrypt_chat_msg做 AES 解密得到明文消息体。第三步如果消息里带media_id还要再调GetMediaData把文件拉下来而且媒体文件是分片的要循环拉直到is_finish为 1。这三步里最容易错的是第一步。私钥必须是生成密钥对时的那把很多人换了私钥或者私钥格式不对PKCS1 和 PKCS8 混用RSA 解密直接返回空。另外encrypt_random_key是 base64 编码的要先解码再解密顺序反了就是一堆乱码。3.2 解密代码与媒体分片下载import base64 from Crypto.PublicKey import RSA from Crypto.Cipher import PKCS1_v1_5, AES # 加载私钥注意是 PKCS1 格式 with open(private_key.pem, rb) as f: private_key RSA.import_key(f.read()) cipher_rsa PKCS1_v1_5.new(private_key) def decrypt_random_key(encrypt_random_key: str) - bytes: 第一步RSA 解密得到中间密钥 encrypted base64.b64decode(encrypt_random_key) # 官方用 PKCS1 v1.5 padding return cipher_rsa.decrypt(encrypted, None) def decrypt_chat_msg(encrypt_chat_msg: str, aes_key: bytes) - str: 第二步AES 解密得到明文 encrypted base64.b64decode(encrypt_chat_msg) # 官方 AES 是 CBC 模式key 长度 32 字节iv 取 key 前 16 字节 iv aes_key[:16] cipher AES.new(aes_key, AES.MODE_CBC, iv) plain cipher.decrypt(encrypted) # 去掉 PKCS7 padding pad_len plain[-1] return plain[:-pad_len].decode(utf-8) # 媒体分片下载 def download_media(handle, sdk, media_id: str, save_path: str): index 0 with open(save_path, wb) as f: while True: buf ctypes.create_string_buffer(1024 * 1024 * 2) ret sdk.GetMediaData(handle, index, media_id, b, b, 30, buf, len(buf)) if ret ! 0: raise RuntimeError(fmedia failed at index {index}) chunk json.loads(buf.value.decode(utf-8)) f.write(base64.b64decode(chunk[data])) if chunk[is_finish] 1: break index 1逻辑说明decrypt_random_key里cipher_rsa.decrypt的第二个参数是 sentinel解密失败会返回它这里传 None 便于判断。decrypt_chat_msg的 AES key 是 32 字节iv 固定取前 16 字节这是官方约定不要自己随机生成 iv。媒体下载的index从 0 开始每次拉一片is_finish为 1 时结束。参数上媒体缓冲给 2MB 够用超时 30 秒。这里有个坑媒体文件不落库只落盘的话后续做检索会很痛苦建议至少把media_id、md5、file_path存一张表。4. 落库设计消息表怎么分、索引怎么建才不拖垮查询4.1 单表还是分表取决于你的日增量会话存档的消息量级差异极大。几十人的小团队一天几千条单表完全够。上千人的销售团队一天几十万条单表三个月就上亿查询会明显变慢。我的经验是日增量超过 10 万条就分表按roomid哈希或者按月分。分表键选roomid的好处是同一个会话的消息落在同一张表查会话上下文快按月分的好处是归档和清理方便。两者可以结合先按月分月内再按 roomid 哈希。索引方面最常用的查询是「某个会话在某段时间的消息」和「某个发送者在某段时间的消息」。所以至少要有(roomid, msg_time)和(from_user, msg_time)两个联合索引。不要给msg_content建全文索引消息体太大全文索引会拖慢写入检索需求用外部搜索引擎解决。4.2 建表 SQL 与入库逻辑CREATE TABLE chat_msg_202501 ( id BIGINT AUTO_INCREMENT PRIMARY KEY, seq BIGINT NOT NULL COMMENT 官方游标序号, msg_id VARCHAR(64) NOT NULL COMMENT 消息唯一id, roomid VARCHAR(128) DEFAULT COMMENT 群会话id单聊为空, from_user VARCHAR(128) NOT NULL COMMENT 发送者, to_list TEXT COMMENT 接收者列表json, msg_type VARCHAR(32) NOT NULL COMMENT text/image/file等, msg_time DATETIME NOT NULL COMMENT 消息时间, content MEDIUMTEXT COMMENT 明文内容, media_id VARCHAR(256) DEFAULT COMMENT 媒体id, media_path VARCHAR(512) DEFAULT COMMENT 媒体落盘路径, create_time DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_msg_id (msg_id), KEY idx_room_time (roomid, msg_time), KEY idx_from_time (from_user, msg_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;入库时用INSERT ... ON DUPLICATE KEY UPDATE或者INSERT IGNORE因为拉取重试会导致同一批消息重复到达msg_id唯一键就是你的后悔药。参数上content用 MEDIUMTEXT 是因为长文本和富文本消息可能超过 TEXT 的 64KB 上限。msg_time一定要用官方返回的时间戳转换不要用入库时间否则排序会乱。import pymysql def save_msg(conn, msg: dict): sql INSERT IGNORE INTO chat_msg_202501 (seq, msg_id, roomid, from_user, to_list, msg_type, msg_time, content, media_id, media_path) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) with conn.cursor() as cur: cur.execute(sql, ( msg[seq], msg[msgid], msg.get(roomid, ), msg[from], json.dumps(msg.get(tolist, [])), msg[msgtype], msg[msgtime], msg.get(text, {}).get(content, ), msg.get(media_id, ), msg.get(media_path, ) )) conn.commit()逻辑说明INSERT IGNORE配合msg_id唯一键天然去重。to_list存 json 字符串方便后续解析。content只取文本内容图片和文件的消息体里没有文本靠media_id关联。参数上msgtime是毫秒时间戳入库前要除以 1000 转成秒。这里注意不同消息类型的字段结构不一样text 在text.contentimage 在image.sdkfileid写解析函数时要按msgtype分支不要硬取一个字段。5. 避坑与排查这五条我踩过你别再踩5.1 现象Init 一直返回非 0日志里没有任何有用信息原因九成是 secret 用错了。会话存档的 secret 是在「管理工具-会话内容存档」里单独生成的不是应用管理的 secret也不是通讯录的 secret。另外 corpid 要和企业一致测试企业拉不到正式企业的数据。解决去管理后台确认 secret 来源用官方提供的调试工具先验证 corpid secret 能通再写代码。如果还是不行检查 sdk 动态库的架构和你的运行环境是否匹配x86 和 arm 的库不能混用。5.2 现象RSA 解密返回 None或者解出来是乱码原因私钥格式不对或者encrypt_random_key没有先 base64 解码。官方生成密钥对时给的是 PKCS1 格式如果你用 openssl 转成了 PKCS8RSA.import_key虽然能读但 padding 行为可能不一致。解决直接用官方给的私钥文件不要转换。解密前先base64.b64decode顺序不能反。如果还是 None打印一下私钥的位数必须是 2048 位。5.3 现象消息拉取一段时间后开始重复或者 seq 跳跃原因游标推进逻辑写错了。常见的是拉取成功后先更新内存里的 seq然后异步落库落库失败时 seq 已经推进下次从新 seq 拉中间那批就丢了。另一种是多个消费者共用一个 seq互相覆盖。解决seq 必须持久化落库和推进游标放在同一个事务里或者至少先落库成功再更新 seq。多消费者场景下用数据库行锁或者分布式锁保证同一时刻只有一个消费者在推进 seq。5.4 现象媒体文件下载到一半失败重试后文件损坏原因媒体分片下载没有做断点续传失败后从头开始但文件已经写了一部分导致内容错位。另外index不是从 0 连续递增的中间某片失败后直接跳到下一片文件就缺了一段。解决下载时先写临时文件全部拉完再 rename 成正式文件。记录已完成的index重试时从断点继续。is_finish为 1 才认为完整不要靠文件大小判断。5.5 现象入库越来越慢查询超时原因单表数据量太大或者索引建错了。常见的是给content建了索引写入时索引维护开销巨大。另一种是msg_time存成了字符串范围查询用不上索引。解决按第 4 章的分表策略拆表msg_time用 DATETIME 类型联合索引的顺序要匹配查询条件的最左前缀。定期用EXPLAIN检查慢查询不要等业务反馈卡了才看。6. 进阶把会话存档接到质检和检索上一个可验证的技巧跑通拉取和落库只是第一步真正让这套源代码产生价值的是下游。我一般会先做一个最小验证随便选一个内部群拉最近 7 天的消息用roomid查出来人工核对消息条数和顺序是否和客户端一致。这个验证能同时暴露游标、解密、入库三个环节的问题比看日志快得多。验证通过后接质检的常见做法是把content同步到 Elasticsearch用msg_id做文档 idroomid和msg_time做过滤字段。这样销售话术检索、敏感词告警都能做。同步时注意不要全量重建用msg_time做增量水位线每次只同步新消息。参数上ES 的refresh_interval设成 30s 而不是默认的 1s写入吞吐能提升不少质检场景对实时性要求没那么高。还有一个容易被忽略的点会话存档的数据是敏感数据落库和落盘都要加密。我一般会在入库前对content做一次 AES 加密密钥走 KMS 管理查询时在应用层解密。这样即使数据库被拖库明文也不会直接泄露。代价是没法用 SQL 直接 like 查询所以检索必须走 ES这也反过来印证了前面说的「不要给 content 建全文索引」。最后说个习惯每次改完拉取或解密逻辑我都会先用一个只有几条消息的测试会话跑一遍确认 seq 从 0 到结束、消息不重不漏、媒体文件能打开再上生产。这个习惯帮我省了至少三次回滚。会话存档这套东西慢就是快游标和密钥这两处稳住了后面都是体力活。希望帮到你。本文还有配套的精品资源点击获取