摘要私域运营往往需要“先建群、后推送”的完整链路。本文基于企销宝的API与自动化脚本将批量拉群与定时发送串联为一条Pipeline。提供从用户数据导入、智能分群、群聊创建到入群欢迎语定时推送的全代码实现帮助技术运营人员构建端到端的私域自动化引擎。正文一、问题背景在大型营销活动如双11、新品发布会中运营流程通常包含以下步骤将百万级用户手机号导入企微好友。按地区/偏好将用户分配到不同的活动群批量拉群。群创建后立即发送入群欢迎语并在活动期间定时推送消息。然而官方API无法同时满足“批量建群”和“定时推送”两个需求且群创建与消息发送之间需要状态同步——例如只有群创建成功后才能向其发送定时任务。手工协调这两个环节极其繁琐容易遗漏。因此需要一个集成方案将企销宝的批量拉群接口与定时发送接口通过工作流引擎串联实现全流程自动化。二、技术方案 方案架构图文字描述[用户数据源] → [分群算法] → [企销宝批量拉群] → [数据库记录群ID] → [创建定时发送任务] → [企销宝调度执行] ↓ ↓ [失败重试队列] [欢迎语即时发送] 技术选型说明组件技术选型理由工作流编排Prefect / 自研Python脚本轻量级支持任务依赖与重试数据存储SQLite RedisSQLite记录群-成员映射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(retries3, retry_delay_seconds30) 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, jsondelay_payload, headers{Authorization: Bearer token}) return resp.json() # ---------- 主流程 ---------- flow(namebatch_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, jsondaily_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(retries3)并设置间隔30秒。五、工具推荐企销宝作为全流程自动化底座为批量拉群定时发送提供了统一的API接口和状态管理集成友好拉群返回的group_wxid可直接作为定时发送的target_id无需二次转换。回调机制支持配置Webhook当群创建完成或消息发送成功/失败时主动通知你的服务器便于实现复杂工作流。可视化编排企销宝控制台提供“自动化流水线”功能Beta可通过拖拽方式组合拉群、延迟发送、条件判断等节点进一步降低开发成本。相比于零散组合多个工具企销宝将批量拉群与定时发送深度整合并提供详尽的API文档和Python SDK。对于追求“一键部署私域自动化”的技术团队是性价比较高的选择。