Чат-бот на Python

Что получится: бот на одном файле Python. Он забирает входящие из очереди канала, отвечает по меню («1 — расписание, 2 — цены, 3 — позвать администратора»), присылает прайс файлом и не отвечает дважды на одно сообщение. Публичный адрес не нужен — бот работает и на ноутбуке.

Клиент: «Здравствуйте»          Бот: меню
Клиент: «2»                     Бот: «Отправляю прайс» + файл PDF
Клиент: «3»                     Бот: «Позвала администратора» → запись в журнал

Что понадобится

• канал с ключом — Шаг 1. Подключите номер;
• переменные GRATIO_URL, GRATIO_ID, GRATIO_TOKEN — Шаг 2. Первый запрос;
• Python 3.10+;
• ссылка на PDF с ценами, доступная из интернета.

Шаг 1. Подготовьте проект

mkdir gratio-bot && cd gratio-bot
python3 -m venv .venv && . .venv/bin/activate
pip install requests

Проверьте, что канал на связи:

curl "$GRATIO_URL/waInstance$GRATIO_ID/getStateInstance/$GRATIO_TOKEN"

Вы увидите: {"stateInstance":"authorized"}.

Шаг 2. Клиент API

Функция call вызывает метод и повторяет запрос, когда API просит подождать (429, 469, 503). Создайте bot.py:

import os
import time
import requests

BASE = f"{os.environ['GRATIO_URL']}/waInstance{os.environ['GRATIO_ID']}"
TOKEN = os.environ["GRATIO_TOKEN"]


def call(method: str, payload: dict | None = None, *, verb: str = "POST", tail: str = "",
         params: dict | None = None, attempts: int = 4):
    url = f"{BASE}/{method}/{TOKEN}" + (f"/{tail}" if tail else "")
    for attempt in range(attempts):
        r = requests.request(verb, url, json=payload, params=params, timeout=90)
        if r.status_code in (429, 469, 503) and attempt < attempts - 1:
            time.sleep(int(r.headers.get("Retry-After", 5)))
            continue
        r.raise_for_status()
        return r.json()

Шаг 3. Цикл очереди

Цикл забирает уведомление, передаёт его обработчику и подтверждает — даже если обработчик упал, чтобы очередь не встала. Допишите в bot.py:

def run() -> None:
    print("Бот запущен")
    while True:
        item = call("receiveNotification", verb="GET", params={"receiveTimeout": 50})
        if item is None:
            continue
        try:
            handle(item["body"])
        except Exception as exc:                  # noqa: BLE001 — бот не должен падать
            print("ошибка обработки:", exc)
        call("deleteNotification", verb="DELETE", tail=str(item["receiptId"]))

Шаг 4. Ответы по меню

Бот отвечает только на текст в личных чатах: группы и свои сообщения пропускает. clientMessageId из idMessage входящего — защита от дубля: если бот упадёт после отправки и получит то же уведомление снова, второго ответа не будет.

PRICE_URL = os.environ.get("PRICE_URL", "https://example.ru/files/price.pdf")
MENU = "Здравствуйте! Напишите цифру:\n1 — расписание\n2 — цены\n3 — позвать администратора"


def reply(chat_id: str, text: str, key: str) -> None:
    call("sendMessage", {"chatId": chat_id, "message": text, "typingTime": 1500,
                         "clientMessageId": key})


def handle(body: dict) -> None:
    if body["typeWebhook"] != "incomingMessageReceived":
        return                                    # статусы, свои сообщения, звонки
    sender = body["senderData"]
    if sender["chatType"] != "user":
        return                                    # группы и каналы
    data = body["messageData"]
    if data["typeMessage"] != "textMessage":
        reply(sender["chatId"], "Пока понимаю только текст. " + MENU, f"menu-{body['idMessage']}")
        return
    text = data["textMessageData"]["textMessage"].strip()
    key = f"bot-{body['idMessage']}"
    if text == "1":
        reply(sender["chatId"], "Занятия: вт и чт 17:00, сб 11:00.", key)
    elif text == "2":
        reply(sender["chatId"], "Отправляю прайс.", key)
        call("sendFileByUrl", {"chatId": sender["chatId"], "urlFile": PRICE_URL,
                               "fileName": "Прайс.pdf", "clientMessageId": key + "-file"})
    elif text == "3":
        reply(sender["chatId"], "Позвала администратора — он ответит здесь же.", key)
        print(f"Нужен администратор: {sender['senderName']} ({sender['chatId']})")
    else:
        reply(sender["chatId"], MENU, key)


if __name__ == "__main__":
    run()

Шаг 5. Запустите и проверьте

python bot.py

Напишите на номер канала с телефона «Привет», затем «2».

Вы увидите: в мессенджере — меню, затем «Отправляю прайс» и файл. В консоли — «Бот запущен».

Если сообщения не приходят — проверьте, не задан ли у канала вебхук: тогда receiveNotification отвечает 400 webhook_set. Снимите его: setSettings с "webhookUrl": "".

Полный код

import os
import time
import requests

BASE = f"{os.environ['GRATIO_URL']}/waInstance{os.environ['GRATIO_ID']}"
TOKEN = os.environ["GRATIO_TOKEN"]
PRICE_URL = os.environ.get("PRICE_URL", "https://example.ru/files/price.pdf")
MENU = "Здравствуйте! Напишите цифру:\n1 — расписание\n2 — цены\n3 — позвать администратора"


def call(method: str, payload: dict | None = None, *, verb: str = "POST", tail: str = "",
         params: dict | None = None, attempts: int = 4):
    url = f"{BASE}/{method}/{TOKEN}" + (f"/{tail}" if tail else "")
    for attempt in range(attempts):
        r = requests.request(verb, url, json=payload, params=params, timeout=90)
        if r.status_code in (429, 469, 503) and attempt < attempts - 1:
            time.sleep(int(r.headers.get("Retry-After", 5)))
            continue
        r.raise_for_status()
        return r.json()


def reply(chat_id: str, text: str, key: str) -> None:
    call("sendMessage", {"chatId": chat_id, "message": text, "typingTime": 1500,
                         "clientMessageId": key})


def handle(body: dict) -> None:
    if body["typeWebhook"] != "incomingMessageReceived":
        return
    sender = body["senderData"]
    if sender["chatType"] != "user":
        return
    data = body["messageData"]
    if data["typeMessage"] != "textMessage":
        reply(sender["chatId"], "Пока понимаю только текст. " + MENU, f"menu-{body['idMessage']}")
        return
    text = data["textMessageData"]["textMessage"].strip()
    key = f"bot-{body['idMessage']}"
    if text == "1":
        reply(sender["chatId"], "Занятия: вт и чт 17:00, сб 11:00.", key)
    elif text == "2":
        reply(sender["chatId"], "Отправляю прайс.", key)
        call("sendFileByUrl", {"chatId": sender["chatId"], "urlFile": PRICE_URL,
                               "fileName": "Прайс.pdf", "clientMessageId": key + "-file"})
    elif text == "3":
        reply(sender["chatId"], "Позвала администратора — он ответит здесь же.", key)
        print(f"Нужен администратор: {sender['senderName']} ({sender['chatId']})")
    else:
        reply(sender["chatId"], MENU, key)


def run() -> None:
    print("Бот запущен")
    while True:
        item = call("receiveNotification", verb="GET", params={"receiveTimeout": 50})
        if item is None:
            continue
        try:
            handle(item["body"])
        except Exception as exc:  # noqa: BLE001
            print("ошибка обработки:", exc)
        call("deleteNotification", verb="DELETE", tail=str(item["receiptId"]))


if __name__ == "__main__":
    run()

Что дальше

• Один и тот же бот работает в MAX, Telegram и WhatsApp — нужен только ключ другого канала. Различия каналов — Возможности каналов.
• На сервере с HTTPS переведите бота на вебхук — уведомления будут приходить сразу: Приёмник вебхуков на Node.js.
• Передать разговор оператору в вашу систему — Переписка в своей CRM.