Crear e integrar VK bot en un grupo a través de VkBotLongPoll [Python]

En este artículo, crearemos un bot y lo integraremos en el grupo VK en Python 3.x

¿Para quién es este artículo?


Para aquellos que desean escribir un bot simple para su comunidad, capaces de identificar equipos y mostrar la respuesta adecuada

Etapas principales



Crear un grupo bot


Comenzaremos creando un bot, es decir, un grupo en VK.

Para hacer esto, vaya a "grupos" → "crear una comunidad".

Seleccione cualquier tipo de comunidad e ingrese un nombre, tema de grupo.

En la página de configuración que se abre, seleccione "Trabajar con la API".

A continuación, debe crear una clave API.

Luego seleccione los parámetros que necesita con acceso para su clave API.

Lo más probable es que tengas que confirmar la acción en VK usando tu teléfono móvil. Luego copie la clave API resultante en algún lugar a un archivo. Aún lo necesitamos.

Entonces necesitas permitir mensajes. Para hacer esto, vaya al "mensaje" y actívelos. También habilite "Funciones de bot" en "Mensajes" -> "Configuración para el bot".

Allí permitiremos agregar una comunidad a grupos si queremos que el bot pueda recibir mensajes de un grupo.

Configurar una encuesta larga


Para trabajar con la API Long Poll, utilizamos la biblioteca vk_api . Puede instalarlo a través de pip.

Antes de trabajar, guardaremos nuestro token API en el archivo config.py, desde allí cargaremos nuestra clave.

Creemos nuestro primer guión. Pongamos nombre a server.py, que será el script principal del servidor.

Importamos los módulos que necesitamos:

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

Crear una clase de servidor:

 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, "-!") 

Ahora cree el archivo server_manager.py, en el que administrará diferentes servidores. Por el momento, para las pruebas, solo escribimos una llamada a la clase Servidor:

 #     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() 

Importante!

El bot solo puede escribir mensajes a los usuarios que le han permitido enviar mensajes. Puedes hacer esto en la página de la comunidad o escribir el bot primero

Si todo se hace correctamente, el bot nos enviará un mensaje personal.

Ahora agregue el bot al grupo y enséñele cómo procesar mensajes.
Para agregar un bot a un grupo, haga clic en "Invitar a una conversación" en el menú derecho de la comunidad.

Agregue la función de inicio al bot, después de lo cual comienza a "escuchar" los mensajes a través de Long Poll (no olvide agregar permisos a los tipos de eventos en la sección "Trabajar con API" -> "Long Poll API" y poner la última versión):

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

Ejecútelo a través de server_manager.py:

 server1.start() 

Ahora, si escribimos un mensaje al grupo, podemos ver el objeto del evento:
<< clase 'vk_api.bot_longpoll.VkBotMessageEvent'> ({'type': 'message_new', 'object': {'date': 1541273151, 'from_id': 25599999999, 'id': 0, 'out': 0, 'peer_id': 2000000001, 'text': '[club172998024 | bot en da Vk] ¡esto es un texto!', 'conversation_message_id': 187, 'fwd_messages': [], 'important': False, 'random_id': 0 , 'archivos adjuntos': [], 'is_hidden': False}, 'group_id': 172998024})>

Además, si escribimos en mensajes privados:
<< clase 'vk_api.bot_longpoll.VkBotMessageEvent'> ({'type': 'message_new', 'object': {'date': 1541273238, 'from_id': 25599999999, 'id': 47, 'out': 0, 'peer_id': 255396611, 'text': 'this is a private message', 'conversation_message_id': 47, 'fwd_messages': [], 'important': False, 'random_id': 0, 'adjuntos': [], 'is_hidden ': False},' group_id ': 172998024})>


A partir de estos datos, debemos prestar atención a type, object.from_id, object.id, object.peer_id, object.text . Los datos recibidos de los mensajes y del grupo no difieren mucho, excepto object.peer_id y object.id .

Si observa detenidamente, object.id para todos los mensajes del grupo es 0, pero no hay mensajes de mensajes personales. Por lo tanto, es posible separar los mensajes recibidos del grupo y de los personales.

Procesamos los datos recibidos dentro de la clase Servidor:

 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'] 


Escribiremos dos mensajes al bot: uno del grupo, uno en PM. Entonces obtenemos:
Nombre de usuario: Arthur
De: San Petersburgo
Texto: [club172998024 | @ club172998024] este es un mensaje del grupo
Tipo: mensaje grupal
-
Nombre de usuario: Arthur
De: San Petersburgo
Texto: este es un mensaje privado
Tipo: mensaje privado
-

Nota


Como puede ver antes del mensaje en el grupo, hay [club172998024 | @ club172998024], para que el equipo procese correctamente, debe deshacerse de todo el contenido entre corchetes o permitir que el bot acceda a toda la correspondencia

Como podemos ver, vk_api nos permite usar fácilmente los métodos de la API VK. Por ejemplo, ahora usamos el método users.get

Una lista de todos los métodos está disponible en: vk.com/dev/methods

Te aconsejo que estudies y experimentes con los métodos que te interesan. Afortunadamente, VK nos proporcionó muy buena documentación, también en ruso.

Para consolidar el material, agreguemos una función para enviar un mensaje a través del método messages.send :

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

<peer_id> es el identificador de destino. Para responder al mensaje de alguien, especificaremos event.object.peer_id como el parámetro peer_id . Es decir, enviaremos un mensaje al lugar de donde vino la solicitud.

Cambiar el método de inicio:

 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},    !") 

Ahora, si el bot recibe el mensaje, nos responderá de la siguiente manera:
Arthur, recibí tu mensaje!

Código completo

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() 


La tarea para arreglar el material:


Cree una función que tome el parámetro peer_id y envíe al usuario una foto cargada a la comunidad. Muelle útil: vk.com/dev/messages.send

Solución
Primero, suba la foto al grupo y ábrala en VK, considere el enlace:
vkcom / club172998024? z = photo-172998024_456239017 % 2Falbum-172998024_256250731

Solo nos interesa la parte resaltada: photo-172998024_456239017 . Pásalo como un argumento para el método messages.send:

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

Agréguelo al método de inicio y obtenga:



Eso es todo lo básico. Lo principal es aprender a usar vk_api usando varios métodos, su lista completa es: vk.com/dev/methods . Si aprende a trabajar con la documentación de la API de VK, puede crear bots de diversa complejidad y propósitos. Un ejemplo de mi bot para un grupo de estudio: github.com/AppLoidx/GroupAssistant/tree/master

Ahora comencemos a crear la lógica del bot.


Cree commander.py, que aceptará comandos y devolverá la respuesta pasada al usuario 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 

Construyamos la arquitectura de nuestro programa:


"Escuchamos" al servidor Long Poll y recibimos un mensaje de usuario ->
Pasamos el mensaje a Commander.input () -> Definir el modo -> Definir el comando ->
Devolvemos la respuesta -> Transferimos al usuario

Para definir el modo y el comando, creamos dos archivos command_enum.py y mode_enum.py. Utilizándolos, definiremos modos y comandos a través de los métodos de la clase 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 


Para cambiar los modos, use [slash ("/") + <mode_name>], y aceptaremos todos los demás comandos como comandos.

Implementamos esto en 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) 


Todo el código está disponible en github: github.com/AppLoidx/VkLongPollBot

Agregar un teclado:


Este es un proceso muy fácil, las complicaciones pueden ser causadas cuando cambiamos el teclado a una firma específica de comandos, que es diferente para cada modo.

Para agregar un teclado al cuadro de diálogo, debe especificar el parámetro de teclado que acepta json en el método messages.send. Se ve así:

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

O puede pasar el teclado directamente desde el archivo .json:

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

Documentación: vk.com/dev/bots_docs_3?f=4.%2BKeyboard%2Bfor%2Bworks

Considere el ejemplo de nuestro programa agregando un teclado.

Primero, cree un archivo 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" }] ] } 

Para quitar el teclado, debe pasar json con botones vacíos:

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

Anule send_message en 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()) 

Y también en el método de inicio:

 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)) 

Como resultado, obtenemos:



Ultima palabra


No debe usar la lista simple de los códigos fuente que se presentan aquí, solo se usan para que comprenda mejor lo que sucede debajo del capó. Por supuesto, todos son utilizables y puede usarlos en partes.

Personalmente, utilicé tal bot para un asistente de grupo que sabía cómo:

  • crear colas a partir de miembros del grupo, incluidos muchos equipos que editaron la cola, como agregar, eliminar, crear, etc.
  • enviar mensajes a todos los participantes
  • preguntas hechas (por ejemplo, en Java)
  • permitió crear aplicaciones para el intercambio de lugares, etc.

Proyecto Github
Fuentes presentadas aquí

Source: https://habr.com/ru/post/es428680/


All Articles