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

《从批量拉群到定时发送：企销宝全流程自动化运营方案》

news 2026/4/20 9:57:19

📌 摘要

私域运营往往需要“先建群、后推送”的完整链路。本文基于企销宝的API与自动化脚本，将批量拉群与定时发送串联为一条Pipeline。提供从用户数据导入、智能分群、群聊创建到入群欢迎语定时推送的全代码实现，帮助技术运营人员构建端到端的私域自动化引擎。

正文

一、问题背景

在大型营销活动（如双11、新品发布会）中，运营流程通常包含以下步骤：

将百万级用户手机号导入企微好友。
按地区/偏好将用户分配到不同的活动群（批量拉群）。
群创建后立即发送入群欢迎语，并在活动期间定时推送消息。

然而，官方API无法同时满足“批量建群”和“定时推送”两个需求，且群创建与消息发送之间需要状态同步——例如，只有群创建成功后才能向其发送定时任务。手工协调这两个环节极其繁琐，容易遗漏。

因此，需要一个集成方案，将企销宝的批量拉群接口与定时发送接口通过工作流引擎串联，实现全流程自动化。

二、技术方案

🧩 方案架构图（文字描述）

[用户数据源] → [分群算法] → [企销宝批量拉群] → [数据库记录群ID] → [创建定时发送任务] → [企销宝调度执行] ↓ ↓ [失败重试队列] [欢迎语即时发送]

🔧 技术选型说明

组件	技术选型	理由
工作流编排	Prefect / 自研Python脚本	轻量级，支持任务依赖与重试
数据存储	SQLite + Redis	SQLite记录群-成员映射，Redis做去重锁
分群逻辑	Pandas + 规则引擎	基于用户标签（如城市、注册时间）动态分配
核心引擎	企销宝	同时提供拉群和定时发送接口，统一认证

⚖️ 与其他方案对比

方案	是否全自动	数据一致性	可观测性	适用场景
手工分步操作	❌	低	无	极小规模
官方API+自研胶水代码	部分	中	需自建	中型活动
企销宝全流程	✅	高（事务日志）	控制台可查	大规模生产级

三、实现步骤

步骤1：环境准备

需要的账号/工具：

企销宝企业版（支持批量拉群和定时发送API）
用户数据表（含手机号、城市、会员等级等字段）
MySQL或SQLite数据库

安装依赖：

pip install pandas sqlalchemy redis prefect # prefect用于工作流

步骤2：功能配置

设计分群规则示例：

# 假设根据城市分群，每个城市一个群，每个群上限150人 group_rules = { "北京": {"group_name": "北京福利群", "max_members": 150}, "上海": {"group_name": "上海福利群", "max_members": 150}, # 其他城市归入“全国群” }

企销宝接口映射：

拉群接口：/group/create（同第一篇）
发送接口：/msg/send（同步）和/task/schedule（定时）

步骤3：代码实现

完整工作流脚本：

import asyncio import pandas as pd from sqlalchemy import create_engine from prefect import flow, task # ---------- 任务1：加载并分群 ---------- @task def load_and_split_users(file_path): df = pd.read_excel(file_path) # 按城市分组 groups_dict = {} for city, group_df in df.groupby('city'): groups_dict[city] = group_df['phone'].tolist() return groups_dict # ---------- 任务2：批量拉群（调用企销宝） ---------- @task(retries=3, retry_delay_seconds=30) async def create_group_for_city(city, members, owner_wxid): # 复用第一篇的create_group函数 group_info = await create_group(owner_wxid, members) return {"city": city, "group_wxid": group_info['group_wxid']} # ---------- 任务3：创建定时发送任务 ---------- @task def schedule_welcome_msg(group_wxid, city): """群创建成功后，立即设置一条10分钟后发送的欢迎语（避免被判定为垃圾）""" import requests payload = { "cron_expr": f"*/10 * * * *", # 此处简化，实际应计算具体时间，例如当前时间+10分钟 "target_type": 1, "target_ids": [group_wxid], "msg_type": 1, "content": f"欢迎加入{city}福利群！晚8点将发送专属优惠券，请留意。" } # 注意：企销宝支持"delay_seconds"参数用于单次延迟发送，更简单 delay_payload = { "target_ids": [group_wxid], "delay_seconds": 600, # 10分钟后发送 "content": f"欢迎加入{city}福利群！..." } resp = requests.post("https://api.qixiaobao.com/v1/msg/delay\_send", json=delay_payload, headers={"Authorization": "Bearer token"}) return resp.json() # ---------- 主流程 ---------- @flow(name="batch_group_and_schedule") async def main_flow(user_file, owner_pool): groups_dict = load_and_split_users(user_file) owner_list = owner_pool # 例如 ["wxid_owner1", "wxid_owner2"] created_groups = [ ] for idx, (city, members) in enumerate(groups_dict.items()): owner = owner_list[idx % len(owner_list)] group_res = await create_group_for_city(city, members, owner) created_groups.append(group_res) # 为每个新群创建定时欢迎语 schedule_welcome_msg(group_res['group_wxid'], city) # 额外：创建活动期间的每日定时推送 for g in created_groups: daily_payload = { "cron_expr": "0 20 * * *", # 每晚8点 "target_ids": [g['group_wxid']], "content": "【每日福利】点击领取红包🧧" } requests.post("https://api.qixiaobao.com/v1/task/schedule", json=daily_payload) print(f"✅ 共创建 {len(created_groups)} 个群，已配置欢迎语及每日推送") if __name__ == "__main__": asyncio.run(main_flow("users.xlsx", ["wxid_a", "wxid_b"]))

运行效果说明：脚本执行后，控制台输出分群统计、每个群的创建耗时、定时任务创建状态。所有群聊和定时任务均可在企销宝后台可视化查看。从导入数据到完成全部配置，100个群约需5分钟（含API调用间隔）。