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

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

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


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

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



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


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

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

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

    На открывшейся странице настроек, выберите «Работа с 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}, я получил ваше сообщение!")

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

    Весь код

    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, 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_manager.py


    # Импортируем созданный нами класс 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

    Решение
    Сначала, загрузим фото в группу и открыв её в ВК рассмотрим ссылку:
    vkcom/club172998024?z=photo-172998024_456239017%2Falbum-172998024_256250731

    Нас интересует лишь выделенная часть: photo-172998024_456239017. Передадим её в качестве аргумента метода messages.send:

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

    Добавим его в метод start и получим:



    Вот и все основы. Главное, научиться пользоваться 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 "Команда не распознана!"

    weather.py
    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
    


    myanimelist.py
    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
    


    yandex_translate.py
    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)
    • давал возможность создавать заявки на обмен мест и т.п.

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

    Похожие публикации

    Комментарии 1
      +2
                          Астрологи объявили неделю ботов для вк.
            Количество постов про написание ботов на питоне увеличено в N раз.

      Только полноправные пользователи могут оставлять комментарии. Войдите, пожалуйста.

      Самое читаемое