30.01.2022 12:58
Jarwix 3 3600 Источник

Создание ботов - довольно заезжанная тема, но все уроки, статьи и различного рода документация дают информацию только о том, как построить бота в один уровень без возможности создания древа из различных всплывающих меню (клавиатур).
А это нужно для создания:

  • Сложных ботов с несколькими уровнями "глубины" (различные меню/клавиатуры)

  • Ботов, созданных одновременно для групповых чатов и для лички сообщества

  • Ботов, с повторяющимися ключевыми командами в различных меню, которые необходимо разделять

Можно сделать многоуровневого бота на Python сразу (без, например, библиотеки threading и без базы данных), но он будет адекватно работать только для одного пользователя в текущий момент времени, потому что одна клавиатура и текущее положение пользователя будет сохраняться в MainThread (основном треде) для всех.

Предварительная подготовка

Устанавливаем MySQL на необходимую машину:

https://dev.mysql.com/downloads/mysql/

Или через терминал (для Linux):

sudo apt install mysql-server

Если у Вас не установлены необходимые библиотеки - то устанавливаем их через терминал:

pip3 install vk_api
pip3 install pymysql

В базе данных нам предварительно нужно создать базу данных в MySQL с названием vktest с таблицей user и полями - iduser и position. Я создаю БД с кодировкой utf8mb4:

CREATE SCHEMA vktest DEFAULT CHARACTER SET utf8mb4 ;
CREATE TABLE `vktest`.`user` (
  `iduser` INT UNSIGNED NOT NULL,
  `position` TINYINT UNSIGNED NULL,
  PRIMARY KEY (`iduser`));

Базовый алгоритм

Основная идея алгоритма в том, что основной тред будет занят слушанием longpoll, который при появлении нового сообщения будет создавать новый тред. В этот новый тред, как аргументы, будут отсылаться айди пользователя и текст сообщения и далее обрабатываться.

В новосозданном треде бот сразу берет текущее положение пользователя в БД, в зависимости от этого положения происходит обработка сообщения и после обновляется положение пользователя в БД на новое и выполняются нужные инструкции.

Для начала представим этот алгоритм для основной функции, которая и будет создавать эти треды (код составлен для сообщений в личку сообщества):

import vk_api
from vk_api import VkUpload
from vk_api.utils import get_random_id
from vk_api.bot_longpoll import VkBotLongPoll, VkBotEventType
import threading
import requests
import random
import pymysql
import pymysql.cursors


if __name__ == '__main__':
    while True:
        session = requests.Session()
        vk_session = vk_api.VkApi(token="%Токен сообщества VK%")
        vk = vk_session.get_api()
        upload = VkUpload(vk_session)
        longpoll = VkBotLongPoll(vk_session, "%ID сообщества VK%")
        try:
            for event in longpoll.listen():
                if event.type == VkBotEventType.MESSAGE_NEW and event.from_user:
                    threading.Thread(target=processing_message, args=(event.obj.from_id, event.obj.text)).start()
        except Exception:
            pass

Весь код в основной функции завернут в while True из-за того, что раньше каждую ночь (возможно также и сейчас) примерно в 4:30 по МСК переставал отвечать VK на запросы и бот падал (видимо перезагружали серверы).

Основная идея тренировочного бота - это сделать три меню, где Основное меню 1 будет иметь возможность попасть в Меню 2 и в Меню 3, а они, в свою очередь, могли вернуться обратно в Основное меню 1:

В формате меню в самом ВКонтакте хочется представить это так:

У нас 3 клавиатуры, которые нужно создать в виде файлов с расширением .json в папке с main.py. Первый файл keyboard_main.json будет с кодом:

keyboard_main.json
{
  "one_time": false,
  "buttons": [
    [{
      "action": {
        "type": "text",
        "label": "Цитаты Дурова"
      },
      "color": "default"
    }],
	[
      {
      "action": {
        "type": "text",
        "label": "Цитаты Цукерберга"
      },
      "color": "default"
    }]
  ]
}

Второй файл keyboard_durov.json с кодом:

keyboard_durov.json
{
  "one_time": false,
  "buttons": [
    [{
      "action": {
        "type": "text",
        "label": "Хочу ещё Дурова"
      },
      "color": "positive"
    }],
	[
      {
      "action": {
        "type": "text",
        "label": "Выйти в главное меню"
      },
      "color": "default"
    }]
  ]
}

Третий, практически идентичный - keyboard_zuckerberg.json:

keyboard_zuckerberg.json
{
  "one_time": false,
  "buttons": [
    [{
      "action": {
        "type": "text",
        "label": "Хочу ещё Цукерберга"
      },
      "color": "positive"
    }],
	[
      {
      "action": {
        "type": "text",
        "label": "Выйти в главное меню"
      },
      "color": "default"
    }]
  ]
}

Значение target у нас processing_message, поэтому создаём функцию в main.py с именем processing_message и c аргументами в виде id юзера и его сообщения:

processing_message
def processing_message(id_user, message_text):
    number_position = take_position(id_user)

    if number_position == 0:
        send_message(id_user, "keyboard_main.json", "Тебя приветствует бот!")
        add_new_line(id_user)

    elif number_position == 1:
        if message_text == "Цитаты Дурова":
            update_position(id_user, "2")
            send_message(id_user, "keyboard_durov.json", durov_quote())
        elif message_text == "Цитаты Цукерберга":
            update_position(id_user, "3")
            send_message(id_user, "keyboard_zuckerberg.json", zuckerberg_quote())
        else:
            send_message(id_user, "keyboard_main.json", "Непонятная команда")

    elif number_position == 2:
        if message_text == "Хочу ещё Дурова":
            send_message(id_user, "keyboard_durov.json", durov_quote())
        elif message_text == "Выйти в главное меню":
            update_position(id_user, "1")
            send_message(id_user, "keyboard_main.json", "Мы в главном меню")
        else:
            send_message(id_user, "keyboard_durov.json", "Непонятная команда")

    elif number_position == 3:
        if message_text == "Хочу ещё Цукерберга":
            send_message(id_user, "keyboard_zuckerberg.json", zuckerberg_quote())
        elif message_text == "Выйти в главное меню":
            update_position(id_user, "1")
            send_message(id_user, "keyboard_main.json", "Мы в главном меню")
        else:
            send_message(id_user, "keyboard_zuckerberg.json", "Непонятная команда")
    
    else:
      	send_message(id_user, "keyboard_main.json", "Произошла какая-то ошибка")

Весь алгоритм этой функции построен на том, что мы сразу берем текущую позицию пользователя и в зависимости от неё обрабатываем сообщение пользователя. Если number_position - это 0 (функция take_position(id_user) вернула 0, то есть в базе данных она не нашла пользователя), то она сразу добавляет его в БД с позицией 1 через функцию add_new_line(id_user), перемещая таким образом его в главное меню и открывая ему клавиатуру keyboard_main.json.

Если number_position от 1 до 3 - то бот выполняет код в соответствующем для него ветвлении if-elif-else.

Функция take_position(id_user) выглядит таким образом:

def take_position(id_user):
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            sql = "SELECT position FROM user WHERE iduser = %s"
            cursor.execute(sql, (id_user))
            line = cursor.fetchone()
            if line is None:
                return_count = 0
            else:
                return_count = line["position"]
    finally:
        connection.close()
    return return_count

В свою очередь add_new_line(id_user) выглядит так:

def add_new_line(id_user):
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            sql = "INSERT INTO user (iduser, position) VALUES (%s, %s)"
            cursor.execute(sql, (id_user, "1"))
        connection.commit()
    finally:
        connection.close()
    return

Когда пользователь имеет свою строку в БД - он уже может приступать к взаимодействию с ботом. И обновление его положения - это функция update_position(id_user, new_position), которая выглядит таким образом:

def update_position(id_user, new_position):
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            sql = "UPDATE user SET position = %s WHERE iduser = %s"
            cursor.execute(sql, (new_position, id_user))
        connection.commit()
    finally:
        connection.close()
    return

Эти три функции не будут работать без основной функции для БД (которую необходимо добавить практически в самое начало, сразу после импорта библиотек) и вписать ваши данные от БД. Я изменяю только название БД и кодировку на те, которые мы создали в самом начале:

def get_connection():
    connection = pymysql.connect(host="%name_host%",
                                 user="%name_user%",
                                 password="%password_user%",
                                 db="vktest",
                                 charset="utf8mb4",
                                 cursorclass=pymysql.cursors.DictCursor)
    return connection

И одна из самых важных функций - отправка сообщения пользователю. Я вывел её в отдельную функцию send_message, так как это нерационально возить такую махину постоянно по всему коду:

def send_message(id_user, id_keyboard, message_text):
    try:
        vk.messages.send(
             user_id=id_user,
             random_id=get_random_id(),
             keyboard=open(id_keyboard, 'r', encoding='UTF-8').read(),
             message=message_text)
    except:
        print("Ошибка отправки сообщения у id" + id_user)

Ничего не забыли? Ах да, нам же нужны крутые цитаты Дурова и Цукерберга:

durov_quote() и zuckerberg_quote()
def durov_quote():
    durov = ['Лучшее решение из возможных — самое простое. И наоборот.',
             'Что такое университет? Это же раздробленная структура с удельными княжествами.',
             'Коммуникация переоценена. Час одиночества продуктивнее недели разговоров.',
             'Проблемы — это спрятанные решения.',
             'Врать вредно для духовной целостности.']
    return random.choice(durov)


def zuckerberg_quote():
    zuckerberg = ['В мире, который меняется очень быстро, единственная стратегия, которая гарантированно '
                  'провальна — не рисковать.',
                  'Двигайтесь быстро и разрушайте препятствия. Если вы ничего не разрушаете, '
                  'Вы движетесь недостаточно быстро.',
                  'Вопрос не в том, что мы хотим знать о человеке. Вопрос стоит так:'
                  '«Что люди хотят рассказать о себе?»',
                  'Люди могут быть очень умными или иметь отличные профессиональные навыки, '
                  'но если они действительно не верят в свое дело, они не будут по-настоящему работать.',
                  'Вопрос, который я задаю себе почти каждый день: сделал ли я самую важную вещь, которую '
                  'я мог бы сделать? Если я не чувствую, что я работаю над самой важной проблемой, где я '
                  'могу помочь, я не буду чувствовать, что хорошо провожу свое время']
    return random.choice(zuckerberg)

Для них мы использовали встроенные списки Python и подключили библиотеку random.

И весь код main.py (целиком) под спойлером для удобства:

main.py
import vk_api
from vk_api import VkUpload
from vk_api.utils import get_random_id
from vk_api.bot_longpoll import VkBotLongPoll, VkBotEventType
import threading
import random
import pymysql
import pymysql.cursors
import requests


def get_connection():
    connection = pymysql.connect(host="%name_host%",
                                 user="%name_user%",
                                 password="%password_user%",
                                 db="vktest",
                                 charset="utf8mb4",
                                 cursorclass=pymysql.cursors.DictCursor)
    return connection


def send_message(id_user, id_keyboard, message_text):
    try:
        vk.messages.send(
             user_id=id_user,
             random_id=get_random_id(),
             keyboard=open(id_keyboard, 'r', encoding='UTF-8').read(),
             message=message_text)
    except:
        print("Ошибка отправки сообщения у id" + id_user)


def add_new_line(id_user):
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            sql = "INSERT INTO user (iduser, position) VALUES (%s, %s)"
            cursor.execute(sql, (id_user, "1"))
        connection.commit()
    finally:
        connection.close()
    return


def take_position(id_user):
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            sql = "SELECT position FROM user WHERE iduser = %s"
            cursor.execute(sql, (id_user))
            line = cursor.fetchone()
            if line is None:
                return_count = 0
            else:
                return_count = line["position"]
    finally:
        connection.close()
    return return_count


def update_position(id_user, new_position):
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            sql = "UPDATE user SET position = %s WHERE iduser = %s"
            cursor.execute(sql, (new_position, id_user))
        connection.commit()
    finally:
        connection.close()
    return


def durov_quote():
    durov = ['Лучшее решение из возможных — самое простое. И наоборот.',
             'Что такое университет? Это же раздробленная структура с удельными княжествами.',
             'Коммуникация переоценена. Час одиночества продуктивнее недели разговоров.',
             'Проблемы — это спрятанные решения.',
             'Врать вредно для духовной целостности.']
    return random.choice(durov)


def zuckerberg_quote():
    zuckerberg = ['В мире, который меняется очень быстро, единственная стратегия, которая гарантированно '
                  'провальна — не рисковать.',
                  'Двигайтесь быстро и разрушайте препятствия. Если вы ничего не разрушаете, '
                  'Вы движетесь недостаточно быстро.',
                  'Вопрос не в том, что мы хотим знать о человеке. Вопрос стоит так:'
                  '«Что люди хотят рассказать о себе?»',
                  'Люди могут быть очень умными или иметь отличные профессиональные навыки, '
                  'но если они действительно не верят в свое дело, они не будут по-настоящему работать.',
                  'Вопрос, который я задаю себе почти каждый день: сделал ли я самую важную вещь, которую '
                  'я мог бы сделать? Если я не чувствую, что я работаю над самой важной проблемой, где я '
                  'могу помочь, я не буду чувствовать, что хорошо провожу свое время']
    return random.choice(zuckerberg)


def processing_message(id_user, message_text):
    number_position = take_position(id_user)

    if number_position == 0:
        send_message(id_user, "keyboard_main.json", "Тебя приветствует бот!")
        add_new_line(id_user)

    elif number_position == 1:
        if message_text == "Цитаты Дурова":
            update_position(id_user, "2")
            send_message(id_user, "keyboard_durov.json", durov_quote())
        elif message_text == "Цитаты Цукерберга":
            update_position(id_user, "3")
            send_message(id_user, "keyboard_zuckerberg.json", zuckerberg_quote())
        else:
            send_message(id_user, "keyboard_main.json", "Непонятная команда")

    elif number_position == 2:
        if message_text == "Хочу ещё Дурова":
            send_message(id_user, "keyboard_durov.json", durov_quote())
        elif message_text == "Выйти в главное меню":
            update_position(id_user, "1")
            send_message(id_user, "keyboard_main.json", "Мы в главном меню")
        else:
            send_message(id_user, "keyboard_durov.json", "Непонятная команда")

    elif number_position == 3:
        if message_text == "Хочу ещё Цукерберга":
            send_message(id_user, "keyboard_zuckerberg.json", zuckerberg_quote())
        elif message_text == "Выйти в главное меню":
            update_position(id_user, "1")
            send_message(id_user, "keyboard_main.json", "Мы в главном меню")
        else:
            send_message(id_user, "keyboard_zuckerberg.json", "Непонятная команда")
    else:
        send_message(id_user, "keyboard_main.json", "Произошла какая-то ошибка")


if __name__ == '__main__':
    while True:
        session = requests.Session()
        vk_session = vk_api.VkApi(token="%Токен сообщества VK%")
        vk = vk_session.get_api()
        upload = VkUpload(vk_session)
        longpoll = VkBotLongPoll(vk_session, "%ID сообщества VK%")
        try:
            for event in longpoll.listen():
                if event.type == VkBotEventType.MESSAGE_NEW and event.from_user:
                    threading.Thread(target=processing_message, args=(event.obj.from_id, event.obj.text)).start()
        except Exception:
            pass

Заключение

Этой статьей я поделился с массами адекватным, как мне кажется, решением для создания многоуровневого бота. Надеюсь, это защитит многих на старте от решений в виде цикла longpoll в цикле longpoll и в ещё одном longpoll, из которых довольно трудно выйти. Всем удачи.

Комментарии (3)


  1. Inspector-Due
    30.01.2022 17:38
    #24009241
    +2

    def take_position(id_user):
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            sql = "SELECT position FROM user WHERE iduser = %s"
            cursor.execute(sql, (id_user))
            line = cursor.fetchone()
            if line is None:
                return_count = 0
            else:
                return_count = line["position"]
    finally:
        connection.close()
    return return_count

    Тут есть одна проблема. Что если достать данные не получится? Тогда return_count будет не объявлен, на последней строчке код ляжет.

    Также if line is None: ... else: ... можно заменить на return_count = line.get("position", 0)

    Честно говоря, не понятен выбор в сторону threads, ведь есть асинхронность.

    Функция processing_message сильно перегружена. И как быть, если функционал (количество состояний) будет расти? К тому же в таком коде легко будет допустить ошибку (как минимум, из-за того, что состояние выражается через числа).

    Ещё мне не понять, почему часть данных находятся в коде (цитаты), а keyboard layout в отдельном файле? К тому же что будет, если вдруг ошибиться в имени файла? (опять ошибка в рантайме)

    Желательно бы вообще вынести все credentials (данные от БД, VK API token...) в константы, а ещё лучше вообще вынести их из кода.

    Также есть вопрос по поводу обработки сообщений. Почему мы обрабатываем текст таким образом: if message_text == "Цитаты Дурова":, хотя есть payload для клавиатур? И что если вдруг текст на кнопках будет одинаков?

    А что если вашим ботом будут пользоваться много людей, придётся очень часто отправлять сообщения, но надо помнить, что у VK API есть лимиты на отправку сообщений (борьба со спамом). Также могут произойти всякие проблемы в сети, надо сделать повторную отправку сообщения. Но её нет. (Кстати, вы сначала обновляете состояние, а только потом отправляете сообщение. Что если сообщение не отправится, но состояние обновится?)


    1. Jarwix Автор
      30.01.2022 18:16
      #24009399
      +2

      Верные замечания, НО здесь базовый код, я не хотел его перегружать проверками.
      Я хотел написать статью по максимально простому пути, чтобы был понятен именно принцип.
      По поводу асинхронности и longpoll - если только aiovk, но даже если объяснять её - то нужно будет объяснить принцип асинхронности изначально. Я не делал такие проекты с этой библиотекой.
      Credentials я не вынес только чтобы не создавать отдельный файл ради трех строк. Так, естественно, правильное замечание.
      По поводу сообщений - в статье я упростил код, но в моих проектах всё стоит в цикле. Я стоплю на полсекунды отправку после ошибки и потом повторяю опять:

      def send_message(id_user, id_keyboard, message_text):
    for i in range(0, 3):
        try:
            vk.messages.send(
                user_id=id_user,
                random_id=get_random_id(),
                keyboard=open(id_keyboard, 'r', encoding='UTF-8').read(),
                message=message_text)
        except:
            print("Ошибка отправки сообщения у id" + str(id_user))
            time.sleep(0.5)
            continue
        break

      Если текст на кнопках будет одинаков, то позиция будет разная.
      А про обновление состояния - даже если сообщение не сработает, то повторное сообщение выведет "Непонятная команда" и выведет правильную клавиатуру.
      Ах да, я сделал это с мыслью что если пользователь успеет кликнуть на сообщение на новой клавиатуре, а позиция ещё не обновится к этому моменту - будет ошибка.

      P.S. Все статьи, которые я читал на хабре с этой темой - нигде так не штудировали код в комментах =)


  1. buvanenko
    01.02.2022 11:34
    #24016185

    Худшее решение, которое можно придумать. Зачем использовать vk_api и многопоточность, катода есть асинхронный vkbottle?