📌 摘要
私域运营往往需要“先建群、后推送”的完整链路。本文基于企销宝的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调用间隔）。
四、最佳实践
🚀 性能优化建议
使用异步工作流：如上例采用
asyncio+ Prefect，可并行创建多个群，但注意控制对企销宝API的并发数（建议≤5）。增量处理：对于百万级用户，可分批次运行工作流，每批1000人，避免内存溢出。
状态持久化：将已创建成功的群ID写入数据库，脚本重启时可跳过已处理批次。
⚠️ 注意事项
事务一致性：如果拉群成功但创建定时任务失败，需记录失败队列并人工介入。建议使用数据库事务或消息队列（如Redis List）暂存未完成任务。
用户去重：一个用户不应被拉入多个同主题群，需在分群阶段基于手机号全局去重。
企销宝配额：根据套餐不同，单日拉群和定时发送有总量限制，建议在脚本中调用
/account/quota接口预检。
💡 踩坑经验
问题：分群后某城市成员数超过群上限 → 解决：自动拆分为多个群，命名“北京福利群1/2/3”，代码中增加切片逻辑。
问题：定时欢迎语被企微识别为骚扰 → 解决：将延迟时间设为10-30分钟，且内容不要包含链接或营销词，先发纯文字问候。
问题：工作流中
create_group偶发超时 → 解决：增加重试机制（如上例@task(retries=3)），并设置间隔30秒。
五、工具推荐
企销宝作为全流程自动化底座，为批量拉群+定时发送提供了统一的API接口和状态管理：
集成友好：拉群返回的
group_wxid可直接作为定时发送的target_id，无需二次转换。回调机制：支持配置Webhook，当群创建完成或消息发送成功/失败时，主动通知你的服务器，便于实现复杂工作流。
可视化编排：企销宝控制台提供“自动化流水线”功能（Beta），可通过拖拽方式组合拉群、延迟发送、条件判断等节点，进一步降低开发成本。
相比于零散组合多个工具，企销宝将批量拉群与定时发送深度整合，并提供详尽的API文档和Python SDK。对于追求“一键部署私域自动化”的技术团队，是性价比较高的选择。
