Создание и интегрирование бота VK в группу через VkBotLongPoll [Python]

В этой статье мы создадим бота и интегрируем его в группу VK на Python 3.x

Для кого эта статья?

Для тех, кто хочет написать простого бота для своего сообщества, умеющего определять команды и выводить соответствующий ответ

Основные стадии

• Для начала создадим бота в ВК, настроим API-ключ и включим Long Poll API
• Настроим Long Poll
• Добавим определения команд, которые будет обрабатывать бот
• Создадим ассоциации команд
• Добавим клавиатуру

Создание группы-бота

Начнем мы с создания бота, а именно группы в ВК.

Для это нужно зайти в «группы» → «создать сообщество».

Выберите любой тип сообщества и введите название, тематику группы.

На открывшейся странице настроек, выберите «Работа с API».

Далее, необходимо создать API-ключ.

Затем выберите нужные вам параметры с доступом для вашего API-ключа.

Скорее всего, вам придётся подтверждать действие в ВК с помощью мобильного телефона. Затем скопируйте полученный API-ключ где-нибудь в файл. Он нам еще понадобится.

Затем нужно разрешить сообщения. Для этого переходим в «сообщения» и включаем их. Также включим «Возможности ботов» в «Сообщения» -> «Настройки для бота».

Там же разрешим добавить сообщество в группы, если мы хотим чтобы бот мог получать сообщения из группы.

Настройка Long Poll

Для работы с Long Poll API используем библиотеку vk_api. Установить его можно через pip.

Перед работой сохраним наш API-токен в файл config.py оттуда мы будем загружать наш ключ.

Создадим наш первый скрипт. Назовем server.py, который будет основным скриптом сервера.

Импортируем нужные нам модули:

import vk_api.vk_api

from vk_api.bot_longpoll import VkBotLongPoll
from vk_api.bot_longpoll import VkBotEventType

Создадим класс-сервер:

class Server:

    def __init__(self, api_token, group_id, server_name: str="Empty"):

        # Даем серверу имя
        self.server_name = server_name

        # Для Long Poll
        self.vk = vk_api.VkApi(token=api_token)
        
        # Для использования Long Poll API
        self.long_poll = VkBotLongPoll(self.vk, group_id)
        
        # Для вызова методов vk_api
        self.vk_api = self.vk.get_api()

    def send_msg(self, send_id, message):
        """
        Отправка сообщения через метод messages.send
        :param send_id: vk id пользователя, который получит сообщение
        :param message: содержимое отправляемого письма
        :return: None
        """
        self.vk_api.messages.send(peer_id=send_id,
                                  message=message)

    def test(self):
        # Посылаем сообщение пользователю с указанным ID
        self.send_msg(255396611, "Привет-привет!")

Теперь создадим файл server_manager.py, в котором будет управлять разными серверами. Пока для тестов пропишем лишь вызов класса Server:

# Импортируем созданный нами класс Server
from server import Server
# Получаем из config.py наш api-token
from config import vk_api_token


server1 = Server(vk_api_token, 172998024, "server1")
# vk_api_token - API токен, который мы ранее создали
# 172998024 - id сообщества-бота
# "server1" - имя сервера

server1.test()

Важно!

Бот может писать сообщения только тем пользователям, которые разрешили боту присылать сообщения. Сделать это можно на странице сообщества или же первым написать боту

Если все сделано правильно, то бот пришлет нам личное сообщение.

Теперь добавим бота в группу и научим его обрабатывать сообщения.
Для добавления бота в группу нужно нажать «Пригласить в беседу» в правом меню сообщества.

Добавим боту функцию start, после вызова которой он начнет «слушать» сообщения через Long Poll (не забудьте добавить разрешения в типы событий в разделе «Работа с API» -> «Long Poll API» и поставить последнюю версию):

def start(self):
        for event in self.long_poll.listen():
            print(event)

Запустим его через server_manager.py:

server1.start()

Теперь, если напишем сообщение в группу, то сможем увидеть объект event'a:

<<class 'vk_api.bot_longpoll.VkBotMessageEvent'>({'type': 'message_new', 'object': {'date': 1541273151, 'from_id': 25599999999, 'id': 0, 'out': 0, 'peer_id': 2000000001, 'text': '[club172998024|bot in da Vk] this is a text!', 'conversation_message_id': 187, 'fwd_messages': [], 'important': False, 'random_id': 0, 'attachments': [], 'is_hidden': False}, 'group_id': 172998024})>

Также, если мы напишем в личные сообщения:

<<class 'vk_api.bot_longpoll.VkBotMessageEvent'>({'type': 'message_new', 'object': {'date': 1541273238, 'from_id': 25599999999, 'id': 47, 'out': 0, 'peer_id': 255396611, 'text': 'это личное сообщение', 'conversation_message_id': 47, 'fwd_messages': [], 'important': False, 'random_id': 0, 'attachments': [], 'is_hidden': False}, 'group_id': 172998024})>

Из этих данных нам следует обратить внимание на type, object.from_id, object.id, object.peer_id, object.text. Данные полученные из сообщений и из группы ничем сильно не отличаются, кроме object.peer_id и object.id.

Если присмотреться, object.id у всех сообщений из группы равен 0, а сообщения из личных нет. Таким образом, можно разделить сообщения получаемые из группы и из личных.

Обработаем полученные данные внутри класса Server:

def start(self):
    for event in self.long_poll.listen():   # Слушаем сервер

        # Пришло новое сообщение
        if event.type == VkBotEventType.MESSAGE_NEW:

            print("Username: " + self.get_user_name(event.object.from_id))
            print("From: " + self.get_user_city(event.object.from_id))
            print("Text: " + event.object.text)
            print("Type: ", end="")
            if event.object.id > 0:
                print("private message")
            else:
                print("group message")
            print(" --- ")

def get_user_name(self, user_id):
    """ Получаем имя пользователя"""
    return self.vk_api.users.get(user_id=user_id)[0]['first_name']

def get_user_city(self, user_id):
    """ Получаем город пользователя"""
    return self.vk_api.users.get(user_id=user_id, fields="city")[0]["city"]['title']

Напишем боту два сообщения: один из группы, один в личку. Тогда получим:

Username: Артур
From: Санкт-Петербург
Text: [club172998024|@club172998024] this is a message from group
Type: group message
—
Username: Артур
From: Санкт-Петербург
Text: this is a private message
Type: private message
—

Примечание

Как вы могли заметить перед сообщением в группе есть [club172998024|@club172998024], для правильной обработки команды следует избавиться от всего содержимого в квадратных скобках, либо разрешить боту доступ ко всей переписке

Как мы видим, vk_api позволяет нам легко использовать методы VK API. К примеру, сейчас мы использовали метод users.get

Список всех методов доступен по ссылке: vk.com/dev/methods

Советую изучить и поэкспериментировать с методами, которые вас заинтересуют. Благо VK предоставило нам очень хорошую документацию, еще и на русском языке.

Чтобы закрепить материал, давайте добавим функцию отправки сообщения через метод messages.send:

def send_message(self, peer_id, message):
        self.vk_api.messages.send(peer_id=peer_id, message=message)

<peer_id> — идентификатор назначения. Чтобы ответить на чье-то сообщение, в качестве параметра peer_id укажем event.object.peer_id. То есть, пошлем сообщение туда, откуда пришел запрос.

Изменим метод start:

def start(self):
    for event in self.long_poll.listen():   # Слушаем сервер

        # Пришло новое сообщение
        if event.type == VkBotEventType.MESSAGE_NEW:

            username = self.get_user_name(event.object.from_id)
            print("Username: " + username)
            print("From: " + self.get_user_city(event.object.from_id))
            print("Text: " + event.object.text)
            print("Type: ", end="")
            if event.object.id > 0:
                print("private message")
            else:
                print("group message")
            print(" --- ")

            self.send_message(event.object.peer_id, f"{username}, я получил ваше сообщение!")

Теперь, если бот примет сообщение, то он нам ответит в таком стиле:

Артур, я получил ваше сообщение!

import vk_api.vk_api

from vk_api.bot_longpoll import VkBotLongPoll
from vk_api.bot_longpoll import VkBotEventType


class Server:

    def __init__(self, api_token, group_id, server_name: str="Empty"):

        # Даем серверу имя
        self.server_name = server_name

        # Для Long Poll
        self.vk = vk_api.VkApi(token=api_token)

        # Для использоания Long Poll API
        self.long_poll = VkBotLongPoll(self.vk, group_id, wait=20)

        # Для вызова методов vk_api
        self.vk_api = self.vk.get_api()

    def send_msg(self, send_id, message):
        """
        Отправка сообщения через метод messages.send
        :param send_id: vk id пользователя, который получит сообщение
        :param message: содержимое отправляемого письма
        :return: None
        """
        self.vk_api.messages.send(peer_id=send_id,
                                  message=message)

    def test(self):
        self.send_msg(255396611, "Привет-привет!")

    def start(self):
        for event in self.long_poll.listen():   # Слушаем сервер

            # Пришло новое сообщение
            if event.type == VkBotEventType.MESSAGE_NEW:

                username = self.get_user_name(event.object.from_id)
                print("Username: " + username)
                print("From: " + self.get_user_city(event.object.from_id))
                print("Text: " + event.object.text)
                print("Type: ", end="")
                if event.object.id > 0:
                    print("private message")
                else:
                    print("group message")
                print(" --- ")

                self.send_message(event.object.peer_id, f"{username}, я получил ваше сообщение!")

    def get_user_name(self, user_id):
        """ Получаем имя пользователя"""
        return self.vk_api.users.get(user_id=user_id)[0]['first_name']

    def get_user_city(self, user_id):
        """ Получаем город пользователя"""
        return self.vk_api.users.get(user_id=user_id, fields="city")[0]["city"]['title']

    def send_message(self, peer_id, message):
        self.vk_api.messages.send(peer_id=peer_id, message=message)

# Импортируем созданный нами класс Server
from server import Server
# Получаем из config.py наш api-token
from config import vk_api_token


server1 = Server(vk_api_token, 172998024, "server1")
server1.start()

Задачка для закрепления материала:

Создайте функцию, которая принимает параметр peer_id и посылает пользователю фотографию загруженную в сообщество. Полезный док: vk.com/dev/messages.send

def send_img(self, peer_id):
    self.vk_api.messages.send(peer_id=peer_id,
                              attachment="photo-172998024_456239017")

Вот и все основы. Главное, научиться пользоваться vk_api, используя различные методы, весь их список: vk.com/dev/methods. Если вы научитесь работать с документацией VK API, то сможете создавать ботов различной сложности и назначений. Пример моего бота для учебной группы: github.com/AppLoidx/GroupAssistant/tree/master

Теперь приступим к созданию логики бота

Создадим commander.py, который будет принимать команды и возвращать ответ, передаваемый пользователю Vk:

class Commander:

    def __init__(self, vk_api, user_id):
        self.vk_api = vk_api
        self.user_id = user_id

    def input(self, msg):
        """
        Функция принимающая сообщения пользователя
        :param msg: Сообщение 
        :return: Ответ пользователю, отправившему сообщение
        """
        pass

Построим архитектуру нашей программы:

«Слушаем» сервер Long Poll и получаем сообщение пользователя ->
Передаем сообщение в Commander.input() -> Определяем режим -> Определяем команду ->
Возвращаем ответ -> Передаем пользователю

Чтобы определить режим и команду, создадим два файла command_enum.py и mode_enum.py. С помощью них мы будем определять режимы и команды через методы класса Enum:

command_enum.py:

from enum import Enum


class Command(Enum):
    """ weather """
    weather = ["weather", "погода"]

    """ myanimelist """
    anime_top = ["top anime", "топ аниме"]

mode_enum.py:

from enum import Enum


class Mode(Enum):
    default = ["Обычный режим", "default"]
    translate = ["Режим переводчика", "translate"]
    get_ans = 0

Для смены режимов используем [слэш("/")+<имя_режима>], а все остальные команды примем как команды.

Реализуем это в Commander.py:

# Перечисления команд, режимов
from command_enum import Command
from mode_enum import Mode

# Рабочие модули
from translate.yandex_translate import Translator
from weather import Weather
from myanimelist import Myanimelist

# Config
from config import yandex_translate_api


class Commander:

    def __init__(self):

        # Текущий, предыдущий режимы
        self.now_mode = Mode.default
        self.last_mode = Mode.default

        self.last_command = None

        # Для запомминания ответов пользователя
        self.last_ans = None

        # Работа с переводом
        self.translator = Translator(yandex_translate_api)

    def change_mode(self, to_mode):
        """
        Меняет режим приема команд
        :param to_mode: Измененный мод
        """
        self.last_mode = self.now_mode
        self.now_mode = to_mode

        self.last_ans = None

    def input(self, msg):
        """
        Функция принимающая сообщения пользователя
        :param msg: Сообщение
        :return: Ответ пользователю, отправившему сообщение
        """

        # Проверка на команду смены мода

        if msg.startswith("/"):
            for mode in Mode:
                if msg[1::] in mode.value:
                    self.change_mode(mode)
                    return "Режим изменен на " + self.now_mode.value[0]
            return "Неизвестный мод " + msg[1::]

        # Режим получения ответа
        if self.now_mode == Mode.get_ans:
            self.last_ans = msg
            self.now_mode = self.last_mode
            return "Ok!"

        if self.now_mode == Mode.default:

            # Погода
            if msg in Command.weather.value:
                return Weather.get_weather_today()

            # Топ аниме
            if msg in Command.anime_top.value:
                res = ""
                top = Myanimelist.get_top()
                for anime in top:
                    res += anime + " : " + top[anime] + "n"

                return res

        if self.now_mode == Mode.translate:
            if self.last_ans is None:

                # Если язык не выбран, просим пользователя ввести
                self.change_mode(Mode.get_ans)
                self.last_command = msg
                return "Выберите язык на который следует перевести"

            elif self.last_ans == "change":

                # Меняем переводимый язык
                self.last_ans = None
                self.change_mode(Mode.default)

            else:
                # Переводим
                return self.translator.translate_to(msg, self.last_ans)

        return "Команда не распознана!"

import requests
from bs4 import BeautifulSoup


class Weather:
    @staticmethod
    def get_weather_today(city: str = "санкт-петербург") -> list:

        http = "https://sinoptik.com.ru/погода-" + city
        b = BeautifulSoup(requests.get(http).text, "html.parser")

        p3 = b.select('.temperature .p3')
        weather1 = p3[0].getText()
        p4 = b.select('.temperature .p4')
        weather2 = p4[0].getText()
        p5 = b.select('.temperature .p5')
        weather3 = p5[0].getText()
        p6 = b.select('.temperature .p6')
        weather4 = p6[0].getText()

        result = ''
        result = result + ('Утром :' + weather1 + ' ' + weather2) + 'n'
        result = result + ('Днём :' + weather3 + ' ' + weather4) + 'n'
        temp = b.select('.rSide .description')
        weather = temp[0].getText()
        result = result + weather.strip()

        return result

import requests
import bs4


class Myanimelist:

    @staticmethod
    def get_top(count: int=5, by: str="") -> dict:
        types = ["", "airing", "upcoming", "tv", "movie", "ova", "special", "bypopularity", "favorite"]
        if by not in types:
            return {"error: ": "Неизвестный тип!"}
        html = requests.get("https://myanimelist.net/topanime.php?type="+by)
        soup = bs4.BeautifulSoup(html.text, "html.parser")

        res = {}

        for anime in soup.select(".ranking-list", limit=count):

            url = anime.select(".hoverinfo_trigger")[0]['href']
            anime = anime.select(".hoverinfo_trigger")[0].findAll("img")[0]
            name = anime['alt'].split(":")[1].strip(" ")
            res[name] = url

        return res

import requests
from config import yandex_translate_api


class Translator:
    """
    Класс-переводчик использующий API Yandex Translate

    Параметры:

    _key -- ключ от API Yandex.Translate
    _yandex_comment -- согласовано с правилами офомления и использования API Yandex.Translate
    """
    def __init__(self, key, comment=None):
        """
        :param key: ключ от API Yandex.Translate
        :param comment: Комментарий к каждому переводу
        """
        self._key = key
        if comment is None:
            self._yandex_comment = "nПереведено сервисом «Яндекс.Переводчик» http://translate.yandex.ru/"
        else:
            self._yandex_comment = comment

    def translate(self, text, lang, to_lang=None):
        """
        Переводит текст с указанного языка в другой указанный

        :param text: Текст, который нужно перевести
        :param lang: исходный язык
        :param to_lang: конечный язык
        :return: Переведенный текст
        """
        if to_lang is not None:
            lang = f"{lang}-{to_lang}"
        main_url = "https://translate.yandex.net/api/v1.5/tr.json/translate"
        response = requests.get(f"{main_url}?"
                                f"key={self._key}&"
                                f"lang={lang}&"
                                f"text={text}")

        return response.json()['text'][0] + self._yandex_comment

    def lang_identify(self, text, hint="ru,en"):
        """
        Идентифицирует язык

        :param text: Текст
        :param hint: Подсказки для определения языка
        :return: код языка
        """
        main_url = "https://translate.yandex.net/api/v1.5/tr.json/detect"
        response = requests.get(f"{main_url}?"
                                f"key={self._key}&"
                                f"hint={hint}&"
                                f"text={text}")

        return response.json()['lang']

    def translate_ru_en(self, text):
        """
        Переводит текст с русского на английский
        :param text: Текст, который нужно перевести
        :return: Текст переведенный на английский язык
        """
        if self.lang_identify(text) == "ru":
            to_lang = "en"
            from_lang = "ru"
        else:
            to_lang = "ru"
            from_lang = "en"

        return self.translate(text, from_lang, to_lang)

    def translate_to_ru(self, text, hint=None):
        """
        Переводит текст на русский

        :param text: Текст, который нужно перевести
        :param hint: Подсказки к определению языка
        :return: Текст переведенный на русский язык
        """
        if hint is None:
            hint = "ru,en"
        from_lang = self.lang_identify(text, hint)

        return self.translate(text, from_lang, "ru")

    def translate_to(self, text, to_lang, hint=None):
        """
        Переводит текст в нужный язык

        :param text: Текст, который нужно перевести
        :param to_lang: Код результирующего языка
        :param hint: Подсказки к определению языка
        :return: Переведенный текст
        """
        if hint is None:
            hint = "ru,en"
        from_lang = self.lang_identify(text, hint)

        return self.translate(text, from_lang, to_lang)

Весь код доступен на гитхабе: github.com/AppLoidx/VkLongPollBot

Добавляем клавиатуру:

Это очень легкий процесс, осложнения могут быть вызваны, когда мы будем изменять клавиатуру под определенную сигнатуру команд, которая отличается для каждого режима.

Чтобы добавить в окне диалогов клавиатуру, необходимо в методе messages.send указать параметр keyboard, принимающий json. Выглядит это следующим образом:

vk_api.messages.send(...,keyboard=keyboard_json,...)

Или же можно передать клавиатуру прямо с файла .json:

vk_api.messages.send(...,keyboard=open(filename,"r",encoding="UTF-8").read()

Документация: vk.com/dev/bots_docs_3?f=4.%2BКлавиатуры%2Bдля%2Bботов

Рассмотрим на примере нашей программы, добавив клавиатуру.

Для начала создадим файл keyboard.json:

{
  "one_time": false,
  "buttons": [
    [{
      "action": {
        "type": "text",
        "label": "top anime"
      },
      "color": "positive"
    },
      {
        "action": {
          "type": "text",
          "label": "weather"
        },
        "color": "positive"
      }],
    [{
      "action": {
        "type": "text",
        "label": "translate"
      },
      "color": "default"
    }]
  ]
}

Чтобы убрать клавиатуры необходимо передать json с пустым buttons:

{"buttons":[],"one_time":true}

Переопределим send_message в server.py:

def send_msg(self, send_id, message):
    """
    Отправка сообщения через метод messages.send
    :param send_id: vk id пользователя, который получит сообщение
    :param message: содержимое отправляемого письма
    :return: None
    """
    return self.vk_api.messages.send(peer_id=send_id,
                                     message=message,
                                     keyboard=open("keyboards/default.json", "r", encoding="UTF-8").read())

И также в методе start:

def start(self):
    for event in self.long_poll.listen():   # Слушаем сервер
        if event.type == VkBotEventType.MESSAGE_NEW:

            if event.object.from_id not in self.users:
                self.users[event.object.from_id] = Commander()

            # Пришло новое сообщение
            if event.type == VkBotEventType.MESSAGE_NEW:

                self.send_msg(event.object.peer_id,
                              self.users[event.object.from_id].input(event.object.text))

В результате получим:

Последнее слово

Не стоит использовать голый листинг исходников представленных здесь, они использованы лишь для того чтобы вы лучше понимали что происходит под капотом. Разумеется, они все юзабельны и можно использовать их по частям.

Лично я использовал такого бота для группового ассистента, который умел:

• создавать очереди из участников группы, в том числе и множество команд редактировавших очередь, таких как добавление, удаление, создание и тд.
• рассылать всем участникам сообщения
• задавал вопросы (например, по языку Java)
• давал возможность создавать заявки на обмен мест и т.п.

Проект на гитхабе
Исходники представленные здесь

Автор: apploid_offical

Источник