# WhatsApp 营销技术实践：从 0 搭建可量化的私域触达系统-编程阁

本文面向有 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_user'wa_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', 'failed'message_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_id=o.user_idWHEREu.segment='high_value'ANDu.wa_opt_in=TRUEANDo.order_dateBETWEENDATE_SUB(NOW(),INTERVAL90DAY)ANDDATE_SUB(NOW(),INTERVAL30DAY)GROUPBYu.user_id,u.phone,u.country_code;

关键原则：把“发给谁”的决策交给 SQL，而不是人工选名单。

五、模板引擎：Jinja2 + 变量校验

用 Jinja2 渲染带变量的消息模板，避免拼接字符串导致的安全和格式问题。

fromjinja2importTemplate template_str="Hi {{ name }}, your {{ product }} is back in stock. Grab it: {{ link }}"template=Template(template_str)payload={"name":"Alice","product":"Pro Plan","link":"https://shop.example.com/restock?u=abc123"}message=template.render(payload)

增加一层校验，防止变量缺失导致发送失败：

required_vars={"name","product","link"}missing=required_vars-payload.keys()ifmissing:raiseValueError(f"Missing template vars:{missing}")

六、发送层：基于 WhatsApp Business API 的批量脚本

下面是简化的发送脚本骨架。生产环境中会加上限流、重试和账号轮换。

importrequestsimporttimefromtypingimportList,Dict WHATSAPP_API_URL="https://graph.facebook.com/v18.0/{phone_number_id}/messages"ACCESS_TOKEN="YOUR_ACCESS_TOKEN"defsend_template_message(phone:str,template_name:str,language:str="en")->Dict:payload={"messaging_product":"whatsapp","recipient_type":"individual","to":phone,"type":"template","template":{"name":template_name,"language":{"code":language}}}headers={"Authorization":f"Bearer{ACCESS_TOKEN}","Content-Type":"application/json"}resp=requests.post(WHATSAPP_API_URL,json=payload,headers=headers)returnresp.json()defbatch_send(users:List[Dict],template_name:str,qps:int=2):interval=1.0/qpsforuserinusers:try:result=send_template_message(user["phone"],template_name)print(f"Sent to{user['phone']}: {result.get('messages', [{}])[0].get('id')}")exceptExceptionase:print(f"Failed to send to{user['phone']}:{e}")time.sleep(interval)

注意：直接调 API 适合有开发资源的团队。如果希望降低维护成本，也可以选择市面上成熟的第三方方案。

七、Webhook 回调：状态回写与风控

配置 Webhook 接收messages和message_status事件：

fromflaskimportFlask,request,jsonify app=Flask(__name__)@app.route("/webhook/whatsapp",methods=["POST"])defwhatsapp_webhook():data=request.jsonforentryindata.get("entry",[]):forchangeinentry.get("changes",[]):value=change.get("value",{})# 消息状态回执statuses=value.get("statuses",[])forstatusinstatuses:update_message_status(message_id=status["id"],status=status["status"],# sent/delivered/read/failedtimestamp=status["timestamp"])# 用户回复内容messages=value.get("messages",[])formsginmessages:store_inbound_message(phone=msg["from"],text=msg.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:returnf"https://shop.example.com/offer?utm_source=whatsapp&utm_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_id=c.campaign_idLEFTJOINwa_attributions aONl.message_id=a.message_idANDa.attributed_atBETWEENl.sent_atANDDATE_ADD(l.sent_at,INTERVAL7DAY)WHEREc.sent_at>=DATE_SUB(NOW(),INTERVAL30DAY)GROUPBYc.campaign_idORDERBYattributed_revenueDESC;