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

ここで、グループにメッセージを書き込むと、イベントオブジェクトが表示されます。
<< class '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':[]、 '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_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からのメッセージの2つをボットに書き込みます。 それから私達は得る:
ユーザー名:アーサー
From:サンクトペテルブルク
テキスト:[club172998024 | @ club172998024]これはグループからのメッセージです
タイプ:グループメッセージ
-
ユーザー名:アーサー
From:サンクトペテルブルク
テキスト:これはプライベートメッセージです
タイプ:プライベートメッセージ
-

ご注意


ご覧のとおり、グループのメッセージの前に[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を指定します。 つまり、リクエストの送信元にメッセージを送信します。

開始方法を変更します。

 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

解決策
まず、写真をグループにアップロードして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 

プログラムのアーキテクチャを構築しましょう:


Long Pollサーバーを「リッスン」し、ユーザーメッセージを受け取ります->
メッセージをCommander.input()に渡します->モードを定義します->コマンドを定義します->
答えを返します->ユーザーに転送します

モードとコマンドを定義するために、command_enum.pyとmode_enum.pyの2つのファイルを作成します。 それらを使用して、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%2Bbots

キーボードを追加して、プログラムの例を考えてみましょう。

まず、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()) 

また、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)) 

その結果、以下が得られます。



最後の言葉


ここに記載されているソースコードの裸のリストは使用しないでください。これらは、内部で何が起こっているかをよりよく理解するためにのみ使用されます。 もちろん、それらはすべて使用可能であり、部分的に使用できます。

個人的に、私はそのようなボットを、次の方法を知っているグループアシスタントに使用しました。


Githubプロジェクト
ここに提示されたソース

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


All Articles