当前位置：首页 > news >正文

基于标签的SOP任务编排——用Celery实现精准定时群发

news 2026/5/12 3:56:52

摘要

无差别群发导致封号风险和高退订率。本文设计了一套基于标签分群的SOP（标准作业程序）系统，通过Python脚本调用企微API，对指定标签下的用户或群进行定时、差异化内容推送。引入CDN加速和延时策略，模拟人工操作，将点击率提升3倍以上。

正文

一、问题背景

运营需对几百个群发布不同活动（如北京群发A活动，上海群发B活动）。手动操作易错且低效。官方群发API限制较多，且对频繁推送有限制。技术手段需解决“精准选群”和“模拟真人发送频率”两大难题。

二、技术方案

任务编排：使用Celery Beat定时任务，从MySQL读取当日SOP计划。
选群逻辑：基于群标签（如region:beijing）筛选chat_id列表。
发送策略：引入延时队列，每发一个群等待5-10秒，避免触发流控。
CDN优化：涉及图片、文件发送时，先上传至CDN获取URL再发送，减少企微服务器带宽压力。

三、实现步骤

步骤1：设计SOP任务表

sql

CREATE TABLE sop_task ( id INT PRIMARY KEY, task_name VARCHAR(255), target_tag VARCHAR(50), -- 群标签，如 ‘region_shanghai’ content_type ENUM(‘text’, ‘image’, ‘link’), content_json TEXT, -- 内容详情 send_time DATETIME, status TINYINT );

步骤2：群发脚本

python

# mass_send.py from celery import Celery import requests import time app = Celery(‘tasks’, broker=‘redis://localhost:6379/0’) @app.task def send_to_groups_by_tag(tag_name, content): # 1. 根据标签获取所有群ID（需要维护群标签关系表） chat_ids = get_chat_ids_by_tag(tag_name) token = get_access_token() for chat_id in chat_ids: # 2. 构造发送体 data = { “chat_id”: chat_id, “msgtype”: “text”, “text”: { “content”: content } } url = f“https://qyapi.weixin.qq.com/cgi-bin/appchat/send?access_token={token}” resp = requests.post(url, json=data) # 3. 频率控制 if resp.json().get(‘errcode’) == 45008: # 频率限制 time.sleep(60) # 暂停1分钟 else: time.sleep(random.randint(3, 8)) # 随机延时 # 记录发送日志 # 注意：需要实现 get_access_token 和 get_chat_ids_by_tag 函数

步骤3：执行定时扫描
另一脚本每分钟扫描sop_task表，将到期的任务丢入Celery队列。