基于企微官方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）

与其他方案对比

三、实现步骤

步骤1：环境准备

需要的账号/工具：

• 企业微信已认证的企业账号，且开通「客户联系」功能

• 自建应用，获取corpid、corpsecret（权限：externalcontact）

• Python 3.9+，安装依赖：

bash

pip install requests apscheduler redis pymysql

• Redis服务（用于限流与任务队列）

配置要求：

• 企业可信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次，设60

• rate：每秒生成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为准。