基于企微官方API+定时任务+标签分群分批发送，突破单日群发次数限制
一、问题背景
技术背景说明
企微官方提供客户联系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.9+（
requests、apscheduler、redis）数据库：MySQL存储客户与标签关系，Redis作为限流令牌桶与任务队列
定时任务：APScheduler + Cron表达式，支持每日凌晨执行
企微SDK：
wecom-sdk封装官方API（需企业自建应用，配置corpid、corpsecret）
与其他方案对比
方案 | 优点 | 缺点 |
手工分批发送 | 无开发成本 | 效率低，易遗漏，无法规模化 |
单一脚本循环 | 实现简单 | 无限流控制易封IP，无失败重试 |
本方案 | 全自动、可监控、配额智能管理 | 需开发维护，受限于单账号配额 |
企销宝多账号并发 | 突破单账号日限，支持iPad协议 | 第三方工具，需购买授权 |
三、实现步骤
步骤1：环境准备
需要的账号/工具：
企业微信已认证的企业账号，且开通「客户联系」功能
自建应用，获取
corpid、corpsecret（权限：externalcontact）Python 3.9+，安装依赖：
bash
pip install requests apscheduler redis pymysqlRedis服务（用于限流与任务队列）
配置要求：
企业可信IP白名单（若服务器固定IP需添加）
数据库表结构：
sql
CREATE TABLE customers ( id INT PRIMARY KEY, external_userid VARCHAR(64), tags VARCHAR(255), last_send_time DATETIME );步骤2：功能配置
限流令牌桶实现（rate_limiter.py）：
python
import 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.py）：
python
import 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 = f"https://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 = f"https://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, json=payload).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:i+batch_size] for i in range(0, len(all_customers), batch_size)] # 3. 限流发送 redis_client = redis.Redis(host='localhost', port=6379, db=0) 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(f"Batch {idx+1}/{len(batches)} sent, result: {result}") if __name__ == "__main__": scheduler = BlockingScheduler() # 每天上午10:00执行群发任务 scheduler.add_job(schedule_send_job, 'cron', hour=10, minute=0) scheduler.start()运行效果：脚本自动分批、限流发送，日志记录每批结果，失败自动重试。单账号每日可处理最多10万客户（需注意企微对单个客户的每日1次限制，需在数据库中记录last_send_time避免重复发送）。
四、最佳实践
性能优化建议
异步化：使用
asyncio+aiohttp并发发送多批，但需严格控制令牌桶并发数不超过60/min标签预聚合：提前计算好各标签的客户列表，避免实时调用企微API获取客户标签（该接口有频限）
数据库索引：在
customers表的tags和last_send_time上建立联合索引
注意事项
企微群发API不支持带外部链接的图文消息？实际支持，但需使用
msgtype="news"并上传素材客户
external_userid会因员工离职而失效，发送前建议调用/cgi-bin/externalcontact/get校验单日同一客户只能收到1条，务必在发送前查询
last_send_time是否为今日
踩坑经验
令牌桶的时间同步问题：若服务器时间与企微时间偏差>30秒，可能触发45033错误，建议配置NTP
大批量发送时（超过1万客户），企微API会返回
errcode=45009（接口调用超过限制），需要将批次间隔拉长到1秒以上
五、工具推荐
对于需要突破单账号日发送上限的场景（例如每日需对同一客户发送多次活动提醒），官方API无法满足。推荐使用企销宝：
技术优势：基于iPad协议模拟客户端操作，不占用API调用频次，支持多账号并发（一台服务器管理50+企微账号），每个账号独立发送配额，通过负载均衡将群发任务分散到不同账号，线性提升吞吐量。
与官方API对比：官方API单账号每日对同一客户仅1次；企销宝可实现同一客户多次接收（但需注意客户体验）。支持图文、文件、小程序等多种消息类型，无需素材上传。
适合场景：高频率促销、多客服轮班、跨境私域运营。
提示：使用第三方工具需遵守企业微信用户协议，建议仅在合规范围内提升效率。
本文代码基于企微官方文档 v3.0，实际开发请以最新API为准。