通过VkBotLongPoll创建VK机器人并将其集成到一个组中[Python]

在本文中,我们将创建一个机器人并将其集成到Python 3.x中的VK组中。

本文适用于谁?


对于那些想要为自己的社区写一个简单的机器人的人,能够确定团队并显示适当的答案

主要阶段



创建一个机器人组


我们将从创建一个机器人(即VK中的组)开始。

为此,请转到“组”→“创建社区”。

选择任何类型的社区,然后输入名称,组主题。

在打开的设置页面上,选择“使用API​​”。

接下来,您需要创建一个API密钥。

然后选择需要访问API密钥的参数。

最有可能的是,您将必须使用手机在VK中确认操作。 然后将生成的API密钥复制到文件中。 我们仍然需要它。

然后,您需要允许消息。 为此,请转到“消息”并打开它们。 同时在“消息”->“机器人的设置”中启用“启动功能”。

如果我们希望漫游器能够接收来自群组的消息,则可以在该群组中添加社区。

设置长轮询


要使用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() 

重要!

该漫游器只能将消息写入允许该漫游器发送消息的用户。 您可以在社区页面上执行此操作,也可以先编写机器人

如果一切操作正确,该机器人将向我们发送一条个人消息。

现在,将该机器人添加到群组中,并教他如何处理消息。
要将漫游器添加到组中,请在社区右侧菜单中单击“邀请进行对话”。

将启动功能添加到机器人中,之后它将开始通过Long Poll“监听”消息(不要忘记在“使用API​​”->“ Long Poll API”部分中为事件类型添加权限并放入最新版本):

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

通过server_manager.py运行它:

 server1.start() 

现在,如果我们向组写一条消息,我们可以看到事件对象:
<<类'vk_api.bot_longpoll.VkBotMessageEvent'>({'type':'message_new','object':{'date':1541273151,'from_id':25599999999,'id':0,'out':0, 'peer_id':2000000001,'text':'[club172998024 |机器人在达Vk]这是一个文本!','conversation_message_id':187,'fwd_messages':[],'重要':False,'random_id':0 ,'attachments':[],'is_hidden':False},'group_id':172998024})>

另外,如果我们在私人消息中写:
<<类'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':[],'重要':False,'random_id':0,'附件':[],'is_hidden ':False},'group_id':172998024})>


从这些数据中,我们应该注意类型object.from_id,object.id,object.peer_id,object.text 。 从消息和从组接收的数据除了object.peer_idobject.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'] 


我们将向机器人发送两条消息:一组消息,一组消息在PM中。 然后我们得到:
用户名:Arthur
来自:圣彼得堡
文字:[club172998024 | @ club172998024]这是来自群组的消息
类型:群组讯息
--
用户名:Arthur
来自:圣彼得堡
文字:这是私人讯息
类型:私人留言
--

注意事项


正如您在组中的消息之前看到的那样,有[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>是目的地标识符。 要回复某人的消息,我们将event.object.peer_id指定为参数peer_id 。 也就是说,我们会将消息发送到请求的来源。

更改启动方法:

 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参数并向用户发送上传到社区的照片的函数。 有用的Dock: vk.com/dev/messages.send

解决方案
首先,将照片上传到群组并在VK中打开,请考虑以下链接:
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 

让我们构建程序的体系结构:


我们“监听”长轮询服务器并收到用户消息->
我们将消息传递给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 


要更改模式,请使用[斜杠(“ /”)+ <mode_name>],我们将接受所有其他命令作为命令。

我们在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上找到: github.com/AppLoidx/VkLongPollBot

添加键盘:


这是一个非常简单的过程,当我们将键盘更改为特定的命令签名时(可能因每种模式而不同),可能会导致复杂化。

要将键盘添加到对话框中,必须在messages.send方法中指定接受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.%2BKeyboard%2Bfor%2Bworks

通过添加键盘来考虑我们程序的示例。

首先,创建一个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":[],"one_time":true} 

覆盖server.py中的send_message:

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

并且在启动方法中:

 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中)
  • 使创建用于交换位置等的应用程序成为可能。

Github项目
此处显示的资源

Source: https://habr.com/ru/post/zh-CN428680/


All Articles