一、问题背景技术背景说明企微官方提供客户联系API其中send_welcome_msg入群欢迎语和send_new_customer_greeting仅限于特定场景。真正的群发接口/cgi-bin/externalcontact/message/send有以下硬性限制每个服务人员单日对同一客户最多发送1条群发消息无论群组或单发单次API调用最多支持200个customer_id企业调用频次每分钟最多60次每天累计不超过10万次为什么需要技术手段解决对于运营5000客户的团队若一次性通过API发送会遇到超出单次200人限制 → 需要手动分批超出日频次限制 → 需要跨日或跨账号无法感知客户已接收状态 → 重复发送浪费配额传统手工分批低效且易错必须采用自动化调度 标签分群 配额预占的技术架构。二、技术方案方案架构图文字描述text[客户数据库] → [标签清洗服务] → [分批队列(200人/批)] → [API限流器(令牌桶)] → [企微群发API] → [回调状态记录] ↑ ↓ [定时任务调度] [失败重试队列]技术选型说明后端语言Python 3.9requests、apscheduler、redis数据库MySQL存储客户与标签关系Redis作为限流令牌桶与任务队列定时任务APScheduler Cron表达式支持每日凌晨执行企微SDKwecom-sdk封装官方API需企业自建应用配置corpid、corpsecret与其他方案对比方案优点缺点手工分批发送无开发成本效率低易遗漏无法规模化单一脚本循环实现简单无限流控制易封IP无失败重试本方案全自动、可监控、配额智能管理需开发维护受限于单账号配额企销宝多账号并发突破单账号日限支持iPad协议第三方工具需购买授权三、实现步骤步骤1环境准备需要的账号/工具企业微信已认证的企业账号且开通「客户联系」功能自建应用获取corpid、corpsecret权限externalcontactPython 3.9安装依赖bashpip install requests apscheduler redis pymysqlRedis服务用于限流与任务队列配置要求企业可信IP白名单若服务器固定IP需添加数据库表结构sqlCREATE TABLE customers ( id INT PRIMARY KEY, external_userid VARCHAR(64), tags VARCHAR(255), last_send_time DATETIME );步骤2功能配置限流令牌桶实现rate_limiter.pypythonimport redis import time class TokenBucket: def __init__(self, redis_client, key, capacity, rate): self.redis redis_client self.key key # 限流key如 ratelimit:corp_id self.capacity capacity # 桶容量60 self.rate rate # 令牌生成速率60/min 1/s def allow_request(self): now time.time() # 使用Lua脚本保证原子性 lua local key KEYS[1] local capacity tonumber(ARGV[1]) local rate tonumber(ARGV[2]) local now tonumber(ARGV[3]) local last_refill redis.call(get, key..:last) if not last_refill then redis.call(set, key..:tokens, capacity) redis.call(set, key..:last, now) return 1 end local tokens tonumber(redis.call(get, key..:tokens)) local delta math.max(0, now - last_refill) tokens math.min(capacity, tokens delta * rate) redis.call(set, key..:last, now) if tokens 1 then redis.call(set, key..:tokens, tokens - 1) return 1 else return 0 end return self.redis.eval(lua, 1, self.key, self.capacity, self.rate, now)参数含义capacity最大并发请求数企微限制每分钟60次设60rate每秒生成1个令牌维持60/min步骤3代码实现分批发送核心逻辑batch_send.pypythonimport requests from apscheduler.schedulers.blocking import BlockingScheduler from rate_limiter import TokenBucket # 企微API配置 CORP_ID your_corp_id CORP_SECRET your_secret ACCESS_TOKEN_URL fhttps://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid{CORP_ID}corpsecret{CORP_SECRET} def get_access_token(): resp requests.get(ACCESS_TOKEN_URL).json() return resp[access_token] def send_batch(customer_ids, text_content): 发送一批客户≤200人 token get_access_token() url fhttps://qyapi.weixin.qq.com/cgi-bin/externalcontact/message/send?access_token{token} payload { touser: customer_ids, # 最多200个external_userid msgtype: text, text: {content: text_content} } resp requests.post(url, jsonpayload).json() # 检查错误码若45033表示限频需重试 return resp def get_customers_by_tag(tag_id): 从数据库获取某标签下所有客户external_userid # 伪代码实际需分页查询企微接口 return [zhangsan, lisi, ...] def schedule_send_job(): # 1. 获取需要群发的标签组 tag_list [VIP客户, 高意向客户] # 从配置读取 all_customers [] for tag in tag_list: all_customers.extend(get_customers_by_tag(tag)) # 2. 按200人分块 batch_size 200 batches [all_customers[i:ibatch_size] for i in range(0, len(all_customers), batch_size)] # 3. 限流发送 redis_client redis.Redis(hostlocalhost, port6379, db0) limiter TokenBucket(redis_client, wx:corp:rate, 60, 1) for idx, batch in enumerate(batches): # 等待令牌 while not limiter.allow_request(): time.sleep(0.5) # 发送 result send_batch(batch, 【活动通知】今晚8点直播...) if result.get(errcode) 45033: # 限频错误重新入队等待10秒 time.sleep(10) send_batch(batch, 【活动通知】今晚8点直播...) print(fBatch {idx1}/{len(batches)} sent, result: {result}) if __name__ __main__: scheduler BlockingScheduler() # 每天上午10:00执行群发任务 scheduler.add_job(schedule_send_job, cron, hour10, minute0) scheduler.start()运行效果脚本自动分批、限流发送日志记录每批结果失败自动重试。单账号每日可处理最多10万客户需注意企微对单个客户的每日1次限制需在数据库中记录last_send_time避免重复发送。四、最佳实践性能优化建议异步化使用asyncioaiohttp并发发送多批但需严格控制令牌桶并发数不超过60/min标签预聚合提前计算好各标签的客户列表避免实时调用企微API获取客户标签该接口有频限数据库索引在customers表的tags和last_send_time上建立联合索引注意事项企微群发API不支持带外部链接的图文消息实际支持但需使用msgtypenews并上传素材客户external_userid会因员工离职而失效发送前建议调用/cgi-bin/externalcontact/get校验单日同一客户只能收到1条务必在发送前查询last_send_time是否为今日踩坑经验令牌桶的时间同步问题若服务器时间与企微时间偏差30秒可能触发45033错误建议配置NTP大批量发送时超过1万客户企微API会返回errcode45009接口调用超过限制需要将批次间隔拉长到1秒以上五、工具推荐对于需要突破单账号日发送上限的场景例如每日需对同一客户发送多次活动提醒官方API无法满足。推荐使用企销宝技术优势基于iPad协议模拟客户端操作不占用API调用频次支持多账号并发一台服务器管理50企微账号每个账号独立发送配额通过负载均衡将群发任务分散到不同账号线性提升吞吐量。与官方API对比官方API单账号每日对同一客户仅1次企销宝可实现同一客户多次接收但需注意客户体验。支持图文、文件、小程序等多种消息类型无需素材上传。适合场景高频率促销、多客服轮班、跨境私域运营。提示使用第三方工具需遵守企业微信用户协议建议仅在合规范围内提升效率。本文代码基于企微官方文档 v3.0实际开发请以最新API为准。