Python e WhatsApp: Automatize Mensagens com API

python -m venv venv
source venv/bin/activate  # Linux/Mac
pip install requests flask python-dotenv

WHATSAPP_TOKEN=seu_token_de_acesso
PHONE_NUMBER_ID=seu_phone_number_id
VERIFY_TOKEN=um_token_secreto_qualquer

import os
import requests
from dotenv import load_dotenv

load_dotenv()

WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN")
PHONE_NUMBER_ID = os.getenv("PHONE_NUMBER_ID")
API_URL = f"https://graph.facebook.com/v19.0/{PHONE_NUMBER_ID}/messages"

def enviar_mensagem_texto(destinatario: str, texto: str) -> dict:
    """Envia uma mensagem de texto simples via WhatsApp."""
    headers = {
        "Authorization": f"Bearer {WHATSAPP_TOKEN}",
        "Content-Type": "application/json",
    }
    payload = {
        "messaging_product": "whatsapp",
        "to": destinatario,
        "type": "text",
        "text": {"body": texto},
    }
    response = requests.post(API_URL, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

# Exemplo de uso
resultado = enviar_mensagem_texto(
    destinatario="5511999999999",
    texto="Olá! Seu pedido #1234 foi confirmado. 🎉"
)
print(resultado)

def enviar_template(destinatario: str, template_name: str, language: str = "pt_BR") -> dict:
    """Envia uma mensagem usando template aprovado."""
    headers = {
        "Authorization": f"Bearer {WHATSAPP_TOKEN}",
        "Content-Type": "application/json",
    }
    payload = {
        "messaging_product": "whatsapp",
        "to": destinatario,
        "type": "template",
        "template": {
            "name": template_name,
            "language": {"code": language},
        },
    }
    response = requests.post(API_URL, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

# Template padrão de teste da Meta
enviar_template("5511999999999", "hello_world")

def enviar_template_com_parametros(
    destinatario: str,
    template_name: str,
    parametros: list[str],
) -> dict:
    """Envia template com variáveis dinâmicas."""
    headers = {
        "Authorization": f"Bearer {WHATSAPP_TOKEN}",
        "Content-Type": "application/json",
    }
    components = [
        {
            "type": "body",
            "parameters": [
                {"type": "text", "text": p} for p in parametros
            ],
        }
    ]
    payload = {
        "messaging_product": "whatsapp",
        "to": destinatario,
        "type": "template",
        "template": {
            "name": template_name,
            "language": {"code": "pt_BR"},
            "components": components,
        },
    }
    response = requests.post(API_URL, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

# Exemplo: template "confirmacao_pedido" com nome e número do pedido
enviar_template_com_parametros(
    "5511999999999",
    "confirmacao_pedido",
    ["Maria", "PED-2026-0331"]
)

from flask import Flask, request, jsonify
import os
from dotenv import load_dotenv

load_dotenv()

app = Flask(__name__)
VERIFY_TOKEN = os.getenv("VERIFY_TOKEN")

@app.route("/webhook", methods=["GET"])
def verificar_webhook():
    """Endpoint de verificação do webhook (challenge da Meta)."""
    mode = request.args.get("hub.mode")
    token = request.args.get("hub.verify_token")
    challenge = request.args.get("hub.challenge")

    if mode == "subscribe" and token == VERIFY_TOKEN:
        return challenge, 200
    return "Forbidden", 403

@app.route("/webhook", methods=["POST"])
def receber_mensagem():
    """Processa mensagens recebidas via webhook."""
    data = request.get_json()

    if data.get("object") == "whatsapp_business_account":
        for entry in data.get("entry", []):
            for change in entry.get("changes", []):
                value = change.get("value", {})
                messages = value.get("messages", [])

                for msg in messages:
                    remetente = msg["from"]
                    tipo = msg["type"]

                    if tipo == "text":
                        texto = msg["text"]["body"]
                        print(f"Mensagem de {remetente}: {texto}")

                        # Responder automaticamente
                        enviar_mensagem_texto(
                            remetente,
                            f"Recebi sua mensagem: '{texto}'. Em breve um atendente responderá!"
                        )

    return jsonify({"status": "ok"}), 200

if __name__ == "__main__":
    app.run(port=5000, debug=True)

import httpx
import asyncio

async def enviar_mensagens_em_lote(destinatarios: list[str], texto: str):
    """Envia mensagens para múltiplos destinatários de forma assíncrona."""
    headers = {
        "Authorization": f"Bearer {WHATSAPP_TOKEN}",
        "Content-Type": "application/json",
    }

    async with httpx.AsyncClient() as client:
        tarefas = []
        for numero in destinatarios:
            payload = {
                "messaging_product": "whatsapp",
                "to": numero,
                "type": "text",
                "text": {"body": texto},
            }
            tarefas.append(client.post(API_URL, headers=headers, json=payload))

        respostas = await asyncio.gather(*tarefas)
        for resp in respostas:
            print(f"Status: {resp.status_code}")

# Executar
asyncio.run(enviar_mensagens_em_lote(
    ["5511999999999", "5521888888888"],
    "Promoção especial para você! 🔥"
))

import time
from functools import wraps

def rate_limiter(max_por_segundo: int = 10):
    """Decorator simples para limitar taxa de envio."""
    intervalo = 1.0 / max_por_segundo

    def decorator(func):
        ultimo_envio = [0.0]

        @wraps(func)
        def wrapper(*args, **kwargs):
            agora = time.time()
            espera = intervalo - (agora - ultimo_envio[0])
            if espera > 0:
                time.sleep(espera)
            ultimo_envio[0] = time.time()
            return func(*args, **kwargs)

        return wrapper
    return decorator

@rate_limiter(max_por_segundo=10)
def enviar_com_limite(destinatario: str, texto: str):
    """Envia mensagem respeitando limite de taxa."""
    return enviar_mensagem_texto(destinatario, texto)

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def criar_sessao_com_retry() -> requests.Session:
    """Cria sessão HTTP com retentativa automática."""
    session = requests.Session()
    retry = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    return session

session = criar_sessao_com_retry()

def enviar_mensagem_robusta(destinatario: str, texto: str) -> dict | None:
    """Envia mensagem com tratamento de erros e retentativas."""
    headers = {
        "Authorization": f"Bearer {WHATSAPP_TOKEN}",
        "Content-Type": "application/json",
    }
    payload = {
        "messaging_product": "whatsapp",
        "to": destinatario,
        "type": "text",
        "text": {"body": texto},
    }
    try:
        response = session.post(API_URL, headers=headers, json=payload)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"Erro HTTP: {e.response.status_code} - {e.response.text}")
    except requests.exceptions.ConnectionError:
        print("Erro de conexão com a API da Meta")
    return None