Как создать бота для WhatsApp на Python в 2026 году

Flask==3.0.0
requests==2.27.1
requests-toolbelt==0.9.1
python-dotenv==0.20.0

pip install -r requirements.txt

API_TOKEN=8FjpwmyFUulh7emXOprrET3xKrwJ984O # API token from your channel
API_URL=https://gate.whapi.cloud # API endpoint URL
PORT=80 # example, 80 or 443
#URL_LINK= The Webhook link to your server {link to server}/hook.

from dotenv import load_dotenv
import os

load_dotenv()
api_token = os.getenv("API_TOKEN")

                                    
import requests

# URL for sending text messages. Can be pulled from .env
url = "https://gate.whapi.cloud/messages/text"

# Data for sending a message
payload = {
    "to": "919984351847",  # Enter the recipient's number in international format
    "body": "Hello! This is a test message."  # Text of message
}

# Headers, including authorization token
headers = {
    "accept": "application/json",
    "content-type": "application/json",
    "authorization": f"Bearer {api_token}"  # Use the token from the .env file
}

# Sending a POST request
response = requests.post(url, json=payload, headers=headers)

# Output server response
print(response.status_code)
print(response.text)
                                    
                                

                                    
from flask import Flask, request, jsonify

app = Flask(__name__)

# Route for processing incoming messages
@app.route('/hook', methods=['POST'])
def webhook():
    # Retrieving data from a request
    data = request.json

    # Logging an incoming message
    print("Received message:", data)

    # Example of incoming message processing
    if "messages" in data:
        for message in data["messages"]:
            sender = message["from"]  # Sender's number
            text = message.get("body", "")  # Text of message

            print(f"Message from {sender}: {text}")

            # Logic for replying to a message
            # Check out our examples on GitHub, where we use more branching on commands for the bot
            if text.lower() == "hello":
                send_response(sender, "Hi there! How can I help you?")
            elif text.lower() == "bye":
                send_response(sender, "Goodbye!")

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

# Function for sending a reply
def send_response(to, body):
    import requests

    url = "https://gate.whapi.cloud/messages/text"
    payload = {
        "to": to,
        "body": body
    }
    headers = {
        "accept": "application/json",
        "content-type": "application/json",
        "authorization": f"Bearer {api_token}"  # Use the token from the .env file
    }

    response = requests.post(url, json=payload, headers=headers)
    print(f"Response to {to}: {response.status_code}, {response.text}")

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=80)

                                    
                                

                                    
import requests

url = "https://gate.whapi.cloud/messages/image"

payload = {
    "to": "919984351847",
    "media": "https://upload.wikimedia.org/wikipedia/en/a/a9/Example.jpg",
    "caption": "An example image"
}
headers = {
    "accept": "application/json",
    "content-type": "application/json",
    "authorization": "Bearer Your_Token"
}

response = requests.post(url, json=payload, headers=headers)

print(response.text)
                                    
                                

                                    
import requests

url = "https://gate.whapi.cloud/messages/document"

payload = {
    "to": "919984351847",
    "media": "data:application/pdf;base64,JVBERi0xLjQKJdPr6eEKMSAwIG9iago8PC9UaXRsZSAoVGVybXMgb2YgU2VydmljZSBXaGFwaS5DbG91ZCkKL0NyZWF0b3IgKE1vemlsbGEvNS4wIFwoV2luZG93cyBOVCAxMC4wOyBXaW42NDsgeDY0XCkgQXBwbGVXZWJLaXQvNTM3LjM2IFwoS0h............",
    "filename": "Terms of Service Whapi.Cloud.pdf",
    "caption": "Hello, I am attaching an important file to my message"
}
headers = {
    "accept": "application/json",
    "content-type": "application/json",
    "authorization": "Bearer Your_Token"
}

response = requests.post(url, json=payload, headers=headers)

print(response.text)
                                    
                                

                                    
import requests

url = "https://gate.whapi.cloud/groups"

payload = {
    "participants": ["919984351847", "919984351848", "919984351849"],
    "subject": "Group Subject 3"
}
headers = {
    "accept": "application/json",
    "content-type": "application/json",
    "authorization": "Bearer Your_Token"
}

response = requests.post(url, json=payload, headers=headers)

print(response.text)
                                    
                                

  
openai>=1.91.0,<2.0.0
httpx>=0.28.1,<1.0.0
httpcore>=1.0.9,<2.0.0
  

                                    
pip install -r requirements.txt
                                

                                    
OPENAI_API_KEY=your_api_key_here
                                

                                    
from flask import Flask, request, jsonify
import requests
from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()
app = Flask(__name__)

API_TOKEN = os.getenv("API_TOKEN")
API_URL = os.getenv("API_URL")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
BACKUP_ADMIN_NUMBER = os.getenv("BACKUP_ADMIN_NUMBER")

if not API_TOKEN or not API_URL or not OPENAI_API_KEY:
    raise RuntimeError("Missing required environment variables in .env")

openai_client = OpenAI(api_key=OPENAI_API_KEY)

# Simple in-memory message deduplication to prevent duplicate processing
processed_messages = set()
MAX_DEDUPLICATION_SIZE = 1000

def send_message(to, body):
    """Sends a text message using the Whapi.Cloud HTTP REST API."""
    headers = {
        "Authorization": f"Bearer {API_TOKEN}",
        "Content-Type": "application/json"
    }
    payload = {"to": to, "body": body}
    response = requests.post(f"{API_URL}/messages/text", json=payload, headers=headers)
    print("Whapi response:", response.status_code, response.text)

def send_backup_alert(error_message):
    """Sends an alert to the backup admin when a critical failure occurs."""
    if BACKUP_ADMIN_NUMBER:
        send_message(BACKUP_ADMIN_NUMBER, f"⚠ CRITICAL BOT ALERT: {error_message}")

def ask_openai(prompt):
    """Sends the user prompt to ChatGPT and returns the text response."""
    response = openai_client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content.strip()

@app.route("/hook/messages", methods=["POST"])
def webhook():
    data = request.json
    print("Incoming webhook:", data)

    for msg in data.get("messages", []):
        # 1. The loop prevention filter: ignore messages sent by the bot itself
        if msg.get("from_me"):
            continue

        msg_id = msg.get("id")
        
        # 2. Webhook message deduplication
        if msg_id in processed_messages:
            print(f"Duplicate message ignored: {msg_id}")
            continue

        processed_messages.add(msg_id)
        if len(processed_messages) > MAX_DEDUPLICATION_SIZE:
            processed_messages.pop()  # Keep memory footprint bounded

        sender = msg.get("chat_id")
        text = (msg.get("text") or {}).get("body", "").strip()

        if text.lower().startswith("/ai "):
            prompt = text[4:].strip()
            if not prompt:
                send_message(sender, "Please provide a prompt after /ai.")
            else:
                try:
                    reply = ask_openai(prompt)
                    send_message(sender, reply)
                except Exception as e:
                    error_msg = f"Failed to process ChatGPT request: {e}"
                    print(error_msg)
                    send_message(sender, "Sorry, I encountered an error processing your request.")
                    send_backup_alert(error_msg)
        else:
            send_message(sender, "Hi! To ask me something, type:\n/ai your question")

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

def register_webhook():
    """Registers the webhook URL with Whapi.Cloud on startup."""
    bot_url = os.getenv("BOT_URL")
    if bot_url:
        headers = {"Authorization": f"Bearer {API_TOKEN}"}
        payload = {
            "webhooks": [
                {
                    "url": bot_url,
                    "events": [{"type": "messages", "method": "post"}],
                    "mode": "method"
                }
            ]
        }
        response = requests.patch(f"{API_URL}/settings", json=payload, headers=headers)
        print("Webhook registration:", response.status_code, response.text)

if __name__ == "__main__":
    register_webhook()
    port = int(os.getenv("PORT", 8080))
    app.run(host="0.0.0.0", port=port)
                                

                                    
gunicorn --workers 3 --bind 0.0.0.0:8080 index:app
                                

                                    
FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8080

CMD ["gunicorn", "--workers", "3", "--bind", "0.0.0.0:8080", "index:app"]
                                

                                    
version: '3.8'

services:
  whatsapp-bot:
    build: .
    ports:
      - "8080:8080"
    env_file:
      - .env
    restart: always