本文面向有 Python/SQL 基础的技术与运营同学分享如何基于 WhatsApp Business API 搭建一套可观测、可回滚的批量营销系统。一、问题定义为什么需要工程化方案某跨境电商团队起初用人工客服在 WhatsApp 上发促销消息。用户量破万后暴露出三个工程问题无法规模化人工发 1 万条消息耗时数天且容易出错。不可归因不知道哪条消息带来了复购订单。无风险控制发送频率、文案质量、账号状态缺乏统一监控。团队决定自建一层营销中台核心目标日发送量 ≥ 5 万条消息级点击归因账号投诉率 0.3%二、整体技术架构┌─────────────┐ ┌──────────────┐ ┌─────────────────┐ │ 订单/用户表 │────▶│ 用户分层服务 │────▶ │ 消息模板引擎 │ │ (MySQL) │ │ (Python) │ │ (Jinja2) │ └─────────────┘ └──────────────┘ └─────────────────┘ │ ┌─────────────┐ ┌──────────────┐ │ │ 效果报表 │◀────│ 回调处理器 │◀────────── ┘ │ (BI/Metabase)│ │ (Webhook) │ └─────────────┘ └──────────────┘核心链路数据层 → 人群圈选 → 模板渲染 → 批量发送 → Webhook 回执 → 数据仓库归因。三、数据层用户标签与订单表设计-- 用户主表CREATETABLEusers(user_idBIGINTPRIMARYKEY,phoneVARCHAR(20)NOTNULL,country_codeVARCHAR(5),segmentVARCHAR(32),-- high_value, churn_risk, new_userwa_opt_inBOOLEANDEFAULTFALSE,created_atTIMESTAMP);-- 消息发送记录表CREATETABLEwa_campaign_logs(log_idBIGINTAUTO_INCREMENTPRIMARYKEY,user_idBIGINT,campaign_idVARCHAR(32),template_nameVARCHAR(64),rendered_messageTEXT,sent_atTIMESTAMP,statusVARCHAR(16),-- sent, delivered, read, failedmessage_idVARCHAR(64),INDEXidx_campaign_sent(campaign_id,sent_at));-- 订单归因表CREATETABLEwa_attributions(order_idBIGINTPRIMARYKEY,user_idBIGINT,campaign_idVARCHAR(32),message_idVARCHAR(64),attributed_revenueDECIMAL(10,2),attributed_atTIMESTAMP);四、人群圈选用 SQL 做分层以“高价值沉默用户”为例SELECTu.user_id,u.phone,u.country_code,MAX(o.order_date)ASlast_order_dateFROMusers uJOINorders oONu.user_ido.user_idWHEREu.segmenthigh_valueANDu.wa_opt_inTRUEANDo.order_dateBETWEENDATE_SUB(NOW(),INTERVAL90DAY)ANDDATE_SUB(NOW(),INTERVAL30DAY)GROUPBYu.user_id,u.phone,u.country_code;关键原则把“发给谁”的决策交给 SQL而不是人工选名单。五、模板引擎Jinja2 变量校验用 Jinja2 渲染带变量的消息模板避免拼接字符串导致的安全和格式问题。fromjinja2importTemplate template_strHi {{ name }}, your {{ product }} is back in stock. Grab it: {{ link }}templateTemplate(template_str)payload{name:Alice,product:Pro Plan,link:https://shop.example.com/restock?uabc123}messagetemplate.render(payload)增加一层校验防止变量缺失导致发送失败required_vars{name,product,link}missingrequired_vars-payload.keys()ifmissing:raiseValueError(fMissing template vars:{missing})六、发送层基于 WhatsApp Business API 的批量脚本下面是简化的发送脚本骨架。生产环境中会加上限流、重试和账号轮换。importrequestsimporttimefromtypingimportList,Dict WHATSAPP_API_URLhttps://graph.facebook.com/v18.0/{phone_number_id}/messagesACCESS_TOKENYOUR_ACCESS_TOKENdefsend_template_message(phone:str,template_name:str,language:stren)-Dict:payload{messaging_product:whatsapp,recipient_type:individual,to:phone,type:template,template:{name:template_name,language:{code:language}}}headers{Authorization:fBearer{ACCESS_TOKEN},Content-Type:application/json}resprequests.post(WHATSAPP_API_URL,jsonpayload,headersheaders)returnresp.json()defbatch_send(users:List[Dict],template_name:str,qps:int2):interval1.0/qpsforuserinusers:try:resultsend_template_message(user[phone],template_name)print(fSent to{user[phone]}: {result.get(messages, [{}])[0].get(id)})exceptExceptionase:print(fFailed to send to{user[phone]}:{e})time.sleep(interval)注意直接调 API 适合有开发资源的团队。如果希望降低维护成本也可以选择市面上成熟的第三方方案。七、Webhook 回调状态回写与风控配置 Webhook 接收messages和message_status事件fromflaskimportFlask,request,jsonify appFlask(__name__)app.route(/webhook/whatsapp,methods[POST])defwhatsapp_webhook():datarequest.jsonforentryindata.get(entry,[]):forchangeinentry.get(changes,[]):valuechange.get(value,{})# 消息状态回执statusesvalue.get(statuses,[])forstatusinstatuses:update_message_status(message_idstatus[id],statusstatus[status],# sent/delivered/read/failedtimestampstatus[timestamp])# 用户回复内容messagesvalue.get(messages,[])formsginmessages:store_inbound_message(phonemsg[from],textmsg.get(text,{}).get(body,))returnjsonify({status:ok})defupdate_message_status(message_id:str,status:str,timestamp:int):# 更新 wa_campaign_logs 表sql UPDATE wa_campaign_logs SET status %s, updated_at FROM_UNIXTIME(%s) WHERE message_id %s execute_sql(sql,(status,timestamp,message_id))基于状态数据可以实时监控送达率、阅读率和失败率。八、归因如何把订单回追到某条消息在消息中带上带 UTM 参数的短链defbuild_tracking_link(user_id:int,campaign_id:str)-str:returnfhttps://shop.example.com/offer?utm_sourcewhatsapputm_campaign{campaign_id}u{user_id}订单落库后用 SQL 做归因窗口分析默认 7 天点击归因SELECTc.campaign_id,COUNT(DISTINCTa.order_id)ASattributed_orders,SUM(a.attributed_revenue)ASattributed_revenue,ROUND(SUM(a.attributed_revenue)/COUNT(DISTINCTl.log_id)*1000,2)ASrpmFROMwa_campaign_logs lJOINwa_campaigns cONl.campaign_idc.campaign_idLEFTJOINwa_attributions aONl.message_ida.message_idANDa.attributed_atBETWEENl.sent_atANDDATE_ADD(l.sent_at,INTERVAL7DAY)WHEREc.sent_atDATE_SUB(NOW(),INTERVAL30DAY)GROUPBYc.campaign_idORDERBYattributed_revenueDESC;九、实际运行效果运行 3 个月后核心指标如下指标基线优化后日发送量800 条/人/天5 万条/天消息打开率人工不可统计71%7 日复购转化率1.2%8.4%账号投诉率0.8%0.18%投诉率下降的关键是分层 错峰发送 失败/退订自动熔断。十、踩坑与建议模板预审核WhatsApp Business API 的模板需提前提交 Meta 审核文案避免促销敏感词。限流与封号新账号先养号初始日发送量控制在 1,000 条以内逐步提升。时区发送根据country_code推断时区避免半夜打扰用户。退订处理用户回复 STOP 后必须立刻写入黑名单并停止发送。十一、何时该用第三方工具如果你的团队没有专职后端开发维护 WhatsApp API、Webhook、账号轮换的成本会很高。这种情况下可以考虑把发送层交给成熟的群发工具自己只保留数据分层和归因部分。我们在对比几款方案时注意到WASender这类工具在变量模板、多账号轮询和分时段发送上做得比较扎实适合想快速跑通 WhatsApp 营销闭环的团队。免责声明文中代码为简化示例生产环境请补充认证、限流、日志和异常处理案例数据已脱敏仅供学习参考。