WhatsApp Automation with Python

import os
import requests

API_VERSION = os.environ.get("WA_API_VERSION", "v21.0")  # use a version Meta currently supports
PHONE_NUMBER_ID = os.environ["WA_PHONE_NUMBER_ID"]
TOKEN = os.environ["WA_ACCESS_TOKEN"]
URL = f"https://graph.facebook.com/{API_VERSION}/{PHONE_NUMBER_ID}/messages"
HEADERS = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}


def _post(payload: dict) -> dict:
    response = requests.post(URL, json=payload, headers=HEADERS, timeout=10)
    response.raise_for_status()
    return response.json()


def send_text(to: str, body: str) -> dict:
    """Free-form text. Only valid inside the 24-hour customer service window."""
    return _post({
        "messaging_product": "whatsapp",
        "to": to,  # digits with country code, for example 38640123456
        "type": "text",
        "text": {"body": body},
    })


def send_template(to: str, name: str, language: str = "en") -> dict:
    """Approved template. Required to start a conversation or after the 24-hour window."""
    return _post({
        "messaging_product": "whatsapp",
        "to": to,
        "type": "template",
        "template": {"name": name, "language": {"code": language}},
    })

import hashlib
import hmac
import os

from flask import Flask, abort, request

from send_message import send_text

app = Flask(__name__)
VERIFY_TOKEN = os.environ["WA_VERIFY_TOKEN"]
APP_SECRET = os.environ["WA_APP_SECRET"]
seen_ids: set[str] = set()  # use Redis or a database in production


@app.get("/webhook")
def verify():
    if (request.args.get("hub.mode") == "subscribe"
            and request.args.get("hub.verify_token") == VERIFY_TOKEN):
        return request.args.get("hub.challenge", ""), 200
    abort(403)


def valid_signature(raw_body: bytes, header: str | None) -> bool:
    if not header or not header.startswith("sha256="):
        return False
    expected = hmac.new(APP_SECRET.encode(), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header.removeprefix("sha256="))


@app.post("/webhook")
def inbound():
    # Verify against the raw bytes, before any JSON parsing.
    if not valid_signature(request.get_data(), request.headers.get("X-Hub-Signature-256")):
        abort(403)

    payload = request.get_json(silent=True) or {}
    for entry in payload.get("entry", []):
        for change in entry.get("changes", []):
            for message in change.get("value", {}).get("messages", []):
                if message["id"] in seen_ids:  # Meta may deliver the same message more than once
                    continue
                seen_ids.add(message["id"])
                handle_message(message)
    return "", 200  # acknowledge quickly; do slow work in a queue


def handle_message(message: dict) -> None:
    sender = message["from"]
    text = message.get("text", {}).get("body", "").strip().lower()

    if text in {"price", "pricing"}:
        send_text(sender, "Our pricing depends on your volume. Reply HUMAN to talk to a person.")
    elif text == "human":
        send_text(sender, "Thanks, a team member will reply shortly.")
        # notify your team here
    else:
        send_text(sender, "Hi! Reply PRICE for pricing or HUMAN to talk to a person.")


if __name__ == "__main__":
    app.run(port=8000)