Création et intégration de bot VK dans un groupe via VkBotLongPoll [Python]

Dans cet article, nous allons créer un bot et l'intégrer au groupe VK dans Python 3.x

À qui s'adresse cet article?


Pour ceux qui veulent écrire un bot simple pour leur communauté, capable d'identifier les équipes et d'afficher la réponse appropriée

Étapes principales



Création d'un groupe de robots


Nous allons commencer par créer un bot, à savoir un groupe en VK.

Pour ce faire, allez dans "groupes" → "créer une communauté".

Sélectionnez n'importe quel type de communauté et entrez un nom, un thème de groupe.

Sur la page des paramètres qui s'ouvre, sélectionnez "Travailler avec l'API".

Ensuite, vous devez créer une clé API.

Sélectionnez ensuite les paramètres dont vous avez besoin avec un accès pour votre clé API.

Très probablement, vous devrez confirmer l'action dans VK en utilisant votre téléphone portable. Copiez ensuite la clé API résultante quelque part dans un fichier. Nous en avons encore besoin.

Ensuite, vous devez autoriser les messages. Pour ce faire, accédez au "message" et activez-les. Activez également "Fonctionnalités du bot" dans "Messages" -> "Paramètres du bot".

Là, nous autoriserons l'ajout d'une communauté à des groupes si nous voulons que le bot puisse recevoir des messages d'un groupe.

Configuration d'un sondage long


Pour travailler avec l'API Long Poll, nous utilisons la bibliothèque vk_api . Vous pouvez l'installer via pip.

Avant de travailler, nous enregistrerons notre jeton API dans le fichier config.py, à partir de là, nous chargerons notre clé.

Créons notre premier script. Appelons server.py, qui sera le script serveur principal.

Nous importons les modules dont nous avons besoin:

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

Créez une classe de serveur:

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

Créez maintenant le fichier server_manager.py, dans lequel il va gérer différents serveurs. Pour le moment, pour les tests, nous écrivons uniquement un appel à la classe 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() 

Important!

Le bot ne peut écrire des messages qu'aux utilisateurs qui ont autorisé le bot à envoyer des messages. Vous pouvez le faire sur la page de la communauté ou écrire d'abord le bot

Si tout est fait correctement, le bot nous enverra un message personnel.

Maintenant, ajoutez le bot au groupe et apprenez-lui à traiter les messages.
Pour ajouter un bot à un groupe, cliquez sur "Inviter à une conversation" dans le menu de droite de la communauté.

Ajoutez la fonction de démarrage au bot, après quoi il commencera à «écouter» les messages via Long Poll (n'oubliez pas d'ajouter des autorisations aux types d'événements dans la section «Travailler avec l'API» -> «Long Poll API» et mettez la dernière version):

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

Exécutez-le via server_manager.py:

 server1.start() 

Maintenant, si nous écrivons un message au groupe, nous pouvons voir l'objet événement:
<< classe '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})>

De plus, si nous écrivons dans des messages privés:
<< classe '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, 'attachments': [], 'is_hidden ': False},' group_id ': 172998024})>


À partir de ces données, nous devons faire attention au type, object.from_id, object.id, object.peer_id, object.text . Les données reçues des messages et du groupe ne diffèrent pas beaucoup sauf object.peer_id et object.id .

Si vous regardez attentivement, object.id pour tous les messages du groupe est 0, mais il n'y a pas de messages de messages personnels. Ainsi, il est possible de séparer les messages reçus du groupe et des messages personnels.

Nous traitons les données reçues à l'intérieur de la classe 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'] 


Nous écrirons deux messages au bot: un du groupe, un en PM. On obtient alors:
Nom d'utilisateur: Arthur
De: Saint-Pétersbourg
Texte: [club172998024 | @ club172998024] ceci est un message du groupe
Type: message de groupe
-
Nom d'utilisateur: Arthur
De: Saint-Pétersbourg
Texte: c'est un message privé
Type: message privé
-

Remarque


Comme vous pouvez le voir avant le message dans le groupe, il y a [club172998024 | @ club172998024], pour que l'équipe traite correctement, vous devez vous débarrasser de tout le contenu entre crochets, ou autoriser le bot à accéder à toute la correspondance

Comme nous pouvons le voir, vk_api nous permet d'utiliser facilement les méthodes de l'API VK. Par exemple, nous avons maintenant utilisé la méthode users.get

Une liste de toutes les méthodes est disponible sur: vk.com/dev/methods

Je vous conseille d'étudier et d'expérimenter les méthodes qui vous intéressent. Heureusement, VK nous a fourni une très bonne documentation, également en russe.

Pour consolider le matériel, ajoutons une fonction pour envoyer un message via la méthode messages.send :

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

<peer_id> est l'identifiant de destination. Pour répondre au message de quelqu'un, nous spécifierons event.object.peer_id comme paramètre peer_id . Autrement dit, nous enverrons un message à l'origine de la demande.

Modifiez la méthode de démarrage:

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

Maintenant, si le bot reçoit le message, il nous répondra dans ce style:
Arthur, j'ai eu ton message!

Code entier

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 tâche de fixation du matériau:


Créez une fonction qui prend le paramètre peer_id et envoie à l'utilisateur une photo téléchargée dans la communauté. Dock utile: vk.com/dev/messages.send

Solution
Tout d'abord, téléchargez la photo dans le groupe et ouvrez-la dans VK, considérez le lien:
vkcom / club172998024? z = photo-172998024_456239017 % 2Falbum-172998024_256250731

Nous ne sommes intéressés que par la partie en surbrillance: photo-172998024_456239017 . Passez-le comme argument à la méthode messages.send:

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

Ajoutez-le à la méthode de démarrage et obtenez:



C'est tout l'essentiel. L'essentiel est d'apprendre à utiliser vk_api en utilisant différentes méthodes, leur liste complète est: vk.com/dev/methods . Si vous apprenez à travailler avec la documentation de l'API VK, vous pouvez créer des bots de complexité et d'objectifs divers. Un exemple de mon bot pour un groupe d'étude: github.com/AppLoidx/GroupAssistant/tree/master

Commençons maintenant à créer la logique du bot.


Créez commander.py, qui acceptera les commandes et renverra la réponse transmise à l'utilisateur 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 

Construisons l'architecture de notre programme:


Nous «écoutons» le serveur Long Poll et recevons un message utilisateur ->
Nous passons le message à Commander.input () -> Définir le mode -> Définir la commande ->
Nous retournons la réponse -> Nous transférons à l'utilisateur

Pour définir le mode et la commande, nous créons deux fichiers command_enum.py et mode_enum.py. En les utilisant, nous définirons les modes et les commandes à travers les méthodes de la classe 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 


Pour changer de mode, utilisez [slash ("/") + <mode_name>], et nous accepterons toutes les autres commandes comme commandes.

Nous implémentons cela dans 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) 


Tout le code est disponible sur github: github.com/AppLoidx/VkLongPollBot

Ajoutez un clavier:


C'est un processus très simple, des complications peuvent être causées lorsque nous changeons le clavier pour une signature spécifique de commandes, qui est différente pour chaque mode.

Pour ajouter un clavier à la boîte de dialogue, vous devez spécifier le paramètre de clavier qui accepte json dans la méthode messages.send. Cela ressemble à ceci:

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

Ou vous pouvez passer le clavier directement à partir du fichier .json:

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

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

Prenons l'exemple de notre programme en ajoutant un clavier.

Créez d'abord un fichier 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" }] ] } 

Pour retirer le clavier, vous devez passer json avec des boutons vides:

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

Remplacez send_message dans 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()) 

Et aussi dans la méthode de démarrage:

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

En conséquence, nous obtenons:



Dernier mot


Vous ne devez pas utiliser la liste nue des codes sources présentés ici, ils ne sont utilisés que pour mieux comprendre ce qui se passe sous le capot. Bien sûr, ils sont tous utilisables et vous pouvez les utiliser en plusieurs parties.

Personnellement, j'ai utilisé un tel bot pour un assistant de groupe qui savait comment:

  • créer des files d'attente à partir des membres du groupe, y compris de nombreuses équipes qui ont modifié la file d'attente, telles que l'ajout, la suppression, la création, etc.
  • envoyer des messages à tous les participants
  • questions posées (par exemple, en Java)
  • a permis de créer des applications pour l'échange de lieux, etc.

Projet Github
Sources présentées ici

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


All Articles