読み飛ばしてください
どうも限界派遣SESのnikawamikanです。
ひとりアドカレ22日目です。
先日、逆翻訳APIを作りました。
なぜAPIで作成したか？というところですが、これはマイクロサービス化をすることによって再利用性を高める狙いがあったからなんですね。（実は何も考えていない）
今回はそのAPIを使ってMisskeyBotを作ってみようと思います。
MisskeyのBOTを作成する
先日作ったAPIといっしょにDiscordのBotを作りましたが、これと同じようにMisskey向けのBotを作っていこうと思います。
まずはMisskeyのAPIを使うためにいくつかの準備が必要です。
ライブラリのインストール
まずはMisskeyのAPIを使うためのライブラリをインストールします。
| ライブラリ | 用途 |
|---|---|
| misskey.py | MisskeyのAPIを使うためのライブラリ |
| websockets | Misskeyインスタンスに対してSocket通信を確立するために利用 |
| aiohttp | 翻訳APIに接続するために利用 |
| aiocache | 翻訳APIの結果をキャッシュするために利用 |
| python-dotenv | 環境変数を読み込むために利用 |
pip install misskey.py websockets aiohttp aiocache python-dotenv
DiscordBotを作ったときのコードを流用して翻訳APIのラッパーを作成
DiscordBotを作ったときのコードを流用して翻訳APIのラッパーを作成します。
直接misskeyのbotにdiscordのbotのコードをimportするのは良くないので、discordとmisskeyのbotで共有して利用できるようにラッパーを作成します。
import aiohttp
from urllib.parse import quote
from aiocache import cached
class ReTranslator:
def __init__(self, url: str):
self.url = url
@cached(600)
async def translate_text(self, text: str, source_lang: str, via_langs: str) -> str:
encoded_text = quote(text)
encoded_source_lang = quote(source_lang)
encoded_via_langs = quote(via_langs)
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.url}/re_translate?text={encoded_text}&source_lang={
encoded_source_lang}&via_langs={encoded_via_langs}"
) as response:
translated_text = await response.text()
# なんか前後に""がついてるので削除
return translated_text
先日のDiscordBotのコードで関数だったものをクラスに変更し、URLを指定できるようにしました。
これにより、APIの接続先が変わったときも柔軟に対応できるようになります。
PYTHONPATHの設定
misskeyのbotから翻訳APIを利用する際は、wrapperを利用しますが、現時点でのディレクトリ構成は以下のようになっています。
.
├── api
│ └── main.py
├── api_wrapper
│ └── re_translate.py
├── bot
│ └── main.py
├── misskey_bot
│ └── main.py
└── requirements.txt
このようになっている場合、misskey_bot/main.pyからapi_wrapper/re_translate.pyをimportすることができません。
そのため、環境変数のPYTHONPATHを設定しておく必要があります。
echo 'export PYTHONPATH=${PYTHONPATH}:/workspaces/re_translator' >> ~/.bashrc
source ~/.bashrc
これはDevContainerを利用している場合は.devcontainer/devcontainer.jsonのpostCreateCommandに追記しておくとコンテナを作成するたびに設定されるようになります。
"postCreateCommand": "echo 'export PYTHONPATH=${PYTHONPATH}:/workspaces/re_translator' >> ~/.bashrc",
これで環境変数が設定されたので、misskey_bot/main.pyからapi_wrapper/re_translate.pyをimportすることができるようになります。
BOTで利用する環境変数を設定
MisskeyBotで利用する環境変数を設定します。
.envファイルを作成し、
BOT_TOKEN="MisskeyのBotのトークン"
MISSKEY_HOST="shahu.ski"
TRANSLATE_API_URL="http://localhost:8000"
のように設定します。
BOT_TOKENですが、これはmisskeyのアクセストークンの発行画面で取得できます。
全ての権限で発行すると、そのユーザーが行えるどのような操作も可能になるため、必要な権限のみを付与したトークンを発行するようにした方が良いです。
そうすることで、不正にトークンを利用された場合でも、そのユーザーが行える操作は制限されるため、被害を最小限に抑えることができます。
ちなみに今回はBOT作成が初めてでどの権限が必要かわからなかったので、全ての権限で取得したのは内緒です。
MisskeyBotを作成
今回はMisskeyのBotに対してメンションが飛んできた時に翻訳APIを利用して翻訳を行い、その結果を返すようにします。
そのため以下のようなフォーマットのメッセージが飛んでくることを想定します。
@bot_name 仰げば 尊し 我が師の恩
教おしえの庭にも はや幾年いくとせ
思えば いと疾とし この年月としつき
今こそ 別れめ いざさらば
このようなメッセージが飛んできた場合、@bot_nameを取り除いたメッセージを翻訳APIに投げ、その結果を返すようにします。
環境変数の読み込み
まずは環境変数を読み込むためのコードを追加します。
これは上記の.envファイルを読み込むためのコードです。
from dotenv import load_dotenv
import os
load_dotenv()
MISSKEY_HOST = os.getenv("MISSKEY_HOST")
BOT_TOKEN = os.getenv("BOT_TOKEN")
TRANSLATE_API_URL = os.getenv("TRANSLATE_API_URL")
if MISSKEY_HOST is None:
raise ValueError("MISSKEY_HOST is not set")
if BOT_TOKEN is None:
raise ValueError("BOT_TOKEN is not set")
if TRANSLATE_API_URL is None:
raise ValueError("TRANSLATE_API_URL is not set")
設定が読み込めない場合はエラーを出力するようにしています。
そうでないと、設定が読み込めない場合に何が原因かわからないためです。
websocket接続をしてメッセージを受信する
まずはwebsocket接続をしてメッセージを受信するコードを書いていきます。
import json
import websockets
from misskey import Misskey
misskey = Misskey(MISSKEY_HOST, i=BOT_TOKEN)
MY_ID = misskey.i()["id"]
WS_URL = f"wss://{MISSKEY_HOST}/streaming?i={BOT_TOKEN}"
async on_mention(note):
pass # ここに処理を書く
async def runner():
async with websockets.connect(WS_URL) as ws:
# websocket接続
await ws.send(json.dumps({
"type": "connect",
"body": {
"channel": "main",
"id": "main"
}
}))
while True:
# メッセージ受信
data = json.loads(s=await ws.recv())
if data['type'] == 'channel':
if data['body']['type'] == 'mention':
note = data['body']['body']
await on_mention(note)
asyncio.get_event_loop().run_until_complete(runner())
ws.sendでconnectを送信することで、websocket接続を確立します。
これにより、ws.recvでメッセージを受信することができるようになります。
今回はメンションのみを気にするため、data['type'] == 'channel'かつdata['body']['type'] == 'mention'の場合のみ処理を行うようにしています。
この辺はPydanticなどで型を定義してあげたほうが良さそうですが、そこそこ大変そうなので今回は省略します。
メンションを受信したときの処理を書く
メンションを受信したときの処理を書いていきます。
from api_wrapper.re_translate import ReTranslator
import re
DEFAULT_SOURCE_LANG = "ja"
DEFAULT_VIA_LANGS = "en,ko,ru,zh"
NOTE_REGEX = re.compile(r'@\S+\s*')
# 翻訳APIのラッパーのインスタンスを作成
re_translator = ReTranslator(TRANSLATE_API_URL)
async def on_mention(note: dict):
if note.get('mentions'):
if MY_ID in note['mentions']:
normalized_text = NOTE_REGEX.sub('', note['text'].strip())
translated_text = await re_translator.translate_text(normalized_text, DEFAULT_SOURCE_LANG, DEFAULT_VIA_LANGS)
misskey.notes_create(text=translated_text, reply_id=note['id'])
on_mention関数はメンションを受信したときの処理を行う関数です。
note.get('mentions')でメンションがあるかどうかを確認し、MY_ID in note['mentions']で自分宛てのメンションかどうかを確認しています。
自分宛てのメンションの場合は、NOTE_REGEX.sub('', note['text'].strip())でメンションを取り除いたテキストを取得し、それを翻訳APIに投げて翻訳結果を取得します。
翻訳結果を取得したら、misskey.notes_create(text=translated_text, reply_id=note['id'])で翻訳結果を返信します。
実行
実行する際は普通にpython main.pyで実行できます。
python main.py
実行して、Misskeyでメンションを飛ばしてみると、翻訳結果が返ってくることが確認できます。
トレーニングガーデンは古いです。
今年も考えてみると、
なにやら続きが気になる文章が帰ってきましたねｗ
まとめ
逆翻訳APIを使ってMisskeyBotを作成してみました。
マイクロサービス化をすることで再利用性を高めることができるので、翻訳のような処理はAPI化しておくと便利ですね。
このAPI化によって、Pythonだけではなく他の言語でも同じようにラッパーを作って利用することができるので、他の言語で作成したBotでも翻訳を利用することができるようになります。
今回のサービスは説明のために一つのリポジトリにまとまっていますが、実際はそれぞれのサービスごとにリポジトリを分けたほうが良いと思います。
そんな感じで少しでも参考になれば幸いです。
それではまた。
参考になった記事