BlueskyAPI+Discordを使ってエゴサBot作ってみた！ REALITY Advent Calendar 2024
REALITY Advent Calendar 17日目担当のサーバーエンジニアの@sola-msrで
す！
この季節、安売りのセールやら忘年会やらイベント盛りだくさんで、気温と同様に懐もめちゃくちゃ寒い状態になっております。助けて！
さて今年は残念ながら（？）コマンドは作っていないのですが、今巷で話題のSNS、mixi2・・・ではなく、BlueskyのAPIを使って色々試してみたので、その紹介をしていこうかなと思います。
（コマンドって何？という方は過去のアドベントカレンダーのnoteなどをご覧ください）
今回はPythonでスクリプトを書いて動かしてみたのと、JavaScript + Google Apps Script(GAS)を使って定期実行させて動かしてみる、ということをやってみたので、今日はその２つについて書いていきたいと思います！
3年前、合宿でX（Twitter）APIを使ってみたことがあるので、本当はその仕組みを使って色々やりたかったのですが、APIがv2になった影響で以前出来ていたことが出来なくなった為、今回は最近よく聞くBlueskyというSNSのAPIを使って、Discordと連携したエゴサbotを作ってみました！
また、今回はDiscordでファンサーバーなどを運営しているライバーの方でも実際に使ってもらえるように、なるべく非エンジニアの方でもわかるように解説しましたので、ご興味ありましたら是非読んで試してみてください！
（※実際に使ってみたいという方は「JavaScript + Google Apps Script(GAS)を使って定期実行させてみる」から読んでもらうと良いかなと思います。）
Pythonでスクリプトを書いて動かしてみる
1. Pythonをインストール
Pythonのインストールについてはここでは割愛させていただきますが、筆者が使用している環境はWindows10でPythonのバージョンは3.13.1です。
（他のOS、バージョンでも問題ないとは思いますが一応載せておきます。）
2. 動かすのに必要なPythonのライブラリをインストール
ここでは2つほど必要なライブラリをインストールしておきます。
まず、Bluesky用のAT Protocolクライアントライブラリ atproto をインストール。
pip install atproto必須ではないですが、コンソールに出力した際に見やすくするために色などを付けるライブラリ colorama をインストールしておきます。
pip install langdetect colorama3. 投稿したいDiscordサーバーのWebhookURLを取得
投稿したいDiscordサーバーのWebhookURLを取得します。
まず、Discordを開いて、既存のサーバーか新規でサーバーを作り、投稿したいチャンネルの右に表示されている歯車のマークから編集画面を開きます。
開いた後に表示されるメニューの「連携サービス」を開き、「ウェブフックを作成」を押して、「Spidey Bot」と作成されたWebhookを編集します。
（名前は好きな名前をつけて大丈夫です。）
後で使用するため、「ウェブフックURLをコピー」を押し、WebhookURLをコピーしてどこかにいったん張り付けておいてください。
（このURLは漏れると大変なので、使用後は破棄するようにしてください。）
4. コードを書いていく
次にスクリプトのコードを書いていきます。
主に使用する検索APIの app.bsky.feed.searchPosts の仕様については以下のページを参考にしてください。
ちなみに、今回作るスクリプトの処理内容としては以下の通りです。
Blueskyの検索APIを叩く
取得した投稿を整形
コンソールにDiscordに送る内容の表示
Discordへ取得した投稿内容の結果を送信
実際のコードはこちら
import requests
from atproto import Client
from colorama import Fore, Style, init
# coloramaの初期化
init(autoreset=True)
# 認証情報の設定
USERNAME = "Blueskyのユーザー名" # 例: example.bsky.social
PASSWORD = "Blueskyのパスワード"
# Discord Webhook URL
DISCORD_WEBHOOK_URL = "投稿したいDiscordサーバーのWebhookURL" # 例: https://discord.com/api/webhooks/**********
# 検索設定
SEARCH_WORD = "#REALITY" # 検索したいワード
LIMIT = 10 # 取得する投稿数
TARGET_LANGUAGE = "ja" # 表示したい言語（ja=日本語）
# 認証トークン取得
def get_access_token():
client = Client()
client.login(USERNAME, PASSWORD)
return client._session.access_jwt
# 投稿内のテキストとURLを結合する関数
def get_full_text(post):
text = post['record']['text']
if 'facets' in post['record']:
for facet in post['record']['facets']:
if 'uri' in facet.get('features', [{}])[0]:
url = facet['features'][0]['uri']
text += f" {url}" # URLをテキストに追加
return text
# Discordへ取得した投稿内容の結果を送信
def post_to_discord(content):
payload = {"content": content}
try:
response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
if response.status_code == 204:
print(f"{Fore.GREEN}Discordに投稿しました: {content}")
else:
print(f"{Fore.RED}Discord投稿エラー: {response.text}")
except Exception as e:
print(f"{Fore.RED}エラーが発生しました: {e}")
# Blueskyから投稿を検索し、Discordに送信
def search_and_post_to_discord(query, lang, limit=30):
url = "https://bsky.social/xrpc/app.bsky.feed.searchPosts" # 検索用エンドポイント
headers = {"Authorization": f"Bearer {get_access_token()}"}
params = {"q": query, "lang": lang, "limit": limit} # リクエストパラメータの設定
try:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
posts = data.get("posts", [])
print(f"\n{Style.BRIGHT}検索結果: '{query}' (言語: '{lang}') をDiscordにポストします...\n")
for post in posts:
author = post['author']['handle']
created_at = post['record'].get('createdAt', 'N/A')
full_text = get_full_text(post)
message = f"**@{author}** ({created_at}):\n{full_text}"
post_to_discord(message)
except requests.exceptions.RequestException as e:
print(f"{Fore.RED}エラーが発生しました: {e}")
# メイン処理
if __name__ == "__main__":
search_and_post_to_discord(SEARCH_WORD, TARGET_LANGUAGE, LIMIT)ここの部分はお好きなように書き換えてください。
SEARCH_WORD = "#REALITY" # 検索したいワード
LIMIT = 10 # 取得する投稿数
TARGET_LANGUAGE = "ja" # 表示したい言語（ja=日本語）5. いざ実行
ファイル名を付けて保存（ファイル名はなんでも大丈夫ですが、ここでは 「main.py」 と名前を付けることにします）し、pythonコマンドで実行してみます。
python main.pyそうするとコンソール部分に投稿内容が表示され、
Discordへ！
うまくBlueskyから投稿を取得し、Discordに通知することが出来ました！
（※今回ユーザーさんが特定できてしまいそうな名前などについては黒塗りで隠させていただいておりますmm）
JavaScript + Google Apps Script(GAS)を使って定期実行させてみる
エンジニアの方でしたら、上記のスクリプトをサーバーなどに上げてcronなどで定期実行させることで、定期的にエゴサbotを動かすことが出来ると思うのですが、今回は非エンジニアの方でも動かすことが出来るようにしていきたいと思います。
具体的には、今回はPythonで書いたコードをJavaScriptに書き換えて、Google Apps Script(GAS) を使い、難しい操作をせずに定期実行できるようにしていきます。
1. Google Apps Script(GAS)で新しいプロジェクトを作成
Google Drive を開き「新規」 → 「その他」 → 「Google Apps Script」の順で開いてください。
この際、作成するプロジェクトの名前はお好きな名前で構いませんが、
あとで見た時に何のスクリプトだったか困らないように、わかりやすい名前にしておくのをオススメします。
2. JavaScriptでスクリプトを書く
Pythonで書いたようなスクリプトをJavaScriptに書き換えたコードがこちら（※完全に全てを置き換えてあるわけではありません）
const BLUESKY_LOGIN_URL = "https://bsky.social/xrpc/com.atproto.server.createSession";
const BLUESKY_API_URL = "https://bsky.social/xrpc/app.bsky.feed.searchPosts";
const DISCORD_WEBHOOK_URL = "投稿したいDiscordサーバーのWebhookURL"; // 例: https://discord.com/api/webhooks/**********
// ユーザー認証情報
const BLUESKY_USERNAME = "Blueskyのユーザー名"; // 例: example.bsky.social
const BLUESKY_PASSWORD = "Blueskyのパスワード";
// Blueskyトークン取得
function getBlueskyAccessToken() {
const payload = {
identifier: BLUESKY_USERNAME,
password: BLUESKY_PASSWORD,
};
const options = {
method: "post",
contentType: "application/json",
payload: JSON.stringify(payload),
};
const response = UrlFetchApp.fetch(BLUESKY_LOGIN_URL, options);
const json = JSON.parse(response.getContentText());
return json.accessJwt;
}
// 投稿テキストとURLを結合する関数
function getFullText(post) {
let text = post.record.text;
// facetsフィールドからURLを取得してテキストに結合
if (post.record.facets) {
post.record.facets.forEach((facet) => {
if (facet.features) {
facet.features.forEach((feature) => {
if (feature.uri) {
text += ` ${feature.uri}`;
}
});
}
});
}
return text;
}
// Bluesky APIから投稿を取得
function fetchBlueskyPosts(query, lang, limit) {
const accessToken = getBlueskyAccessToken();
const headers = { Authorization: `Bearer ${accessToken}` };
const options = {
method: "get",
headers: headers,
};
// 各パラメータを設定
const url = `${BLUESKY_API_URL}?q=${encodeURIComponent(query)}&lang=${lang}&limit=${limit}`;
const response = UrlFetchApp.fetch(url, options);
const json = JSON.parse(response.getContentText());
return json.posts || [];
}
// Discordに投稿を送る関数
function sendToDiscord(message) {
const payload = { content: message };
const options = {
method: "post",
contentType: "application/json",
payload: JSON.stringify(payload),
muteHttpExceptions: true, // レートリミットエラー取得
};
while (true) {
const response = UrlFetchApp.fetch(DISCORD_WEBHOOK_URL, options);
const statusCode = response.getResponseCode();
if (statusCode === 429) {
const retryAfter = JSON.parse(response.getContentText()).retry_after || 1;
console.warn(`レートリミット発生、${retryAfter}秒後に再試行します...`);
Utilities.sleep(retryAfter * 1000); // レートリミット待機
} else {
console.log("Discordに投稿しました:", message);
Utilities.sleep(1000); // 1秒待機（レートリミット回避）
break;
}
}
}
// メイン処理（指定言語の投稿のみ取得してDiscordに送信）
function main() {
const searchWord = "#REALITY"; // 検索キーワード
const lang = "ja"; // 言語指定: 日本語
const limit = 10; // 取得する投稿数
console.log(`Blueskyで「${searchWord}」（言語: ${lang}）を検索中...`);
const posts = fetchBlueskyPosts(searchWord, lang, limit);
console.log(`日本語の投稿が${posts.length}件見つかりました。`);
posts.forEach((post) => {
const author = post.author.handle;
const createdAt = post.record.createdAt || "N/A";
const fullText = getFullText(post);
const message = `**@${author}** (${createdAt}):\n${fullText}`;
sendToDiscord(message);
});
console.log("処理が完了しました。");
}ここの部分はお好きなように書き換えてください。
const searchWord = "#REALITY"; // 検索キーワード
const lang = "ja"; // 言語指定: 日本語
const limit = 10; // 取得する投稿数以下はコードを貼り付けたイメージです。
3. 実行する関数を「main」にしておく
実行する前に、デバッグと書いてあるボタンの横のセレクトボックスを押して、「main」に設定しておいてください。
4. 動くか実行してみる
定期実行の設定をする前にまず一度「実行」を押して実際に動くか確認しておきましょう。
この時、以下のようなダイアログが出ると思いますが、「権限を確認」を押して続行させます。
そうすると、次のダイアログが表示されますが、ここも「詳細」を開いて「プロジェクト名（安全ではないページ）に移動」を押してください。
また、以下のようなダイアログが出ますが、ここも「許可」を押して進めて大丈夫です。
ここまで済ませると、実行していたスクリプトが動き出すと思うので、実行ログを眺めてエラーが出ていないかを確認します。
GAS側で正常に動いていることを確認したら、Discordにもちゃんと送信されているかも確認しておいてください。
5. 定期実行の設定を行う
最後に定期実行の設定をしていきましょう！
まず、左のメニューから「トリガー」を開いてください。
右下にある「トリガーを追加」を押すと、以下のようなダイアログが出ると思うので、「実行する関数を選択」を 「main」に選択し、「時間の間隔を選択（時間）」を好きなように設定し「保存」を押します。
（分からなければその他の項目はそのままで大丈夫です。）
設定したトリガーが画面に表示されていればOKです！
これで定期的にBlueskyで検索したいワードについて検索し、自動で定期的に投稿内容をDiscordに通知してくれるようになりました！
おわりに
まだまだBlueskyAPIで色々と出来そうなので、また機会があれば触っていきたいと思います。
X（Twitter）もいいですが、Blueskyももっと盛り上がっていくといいですね！
明日はようてんさんの「REALITY x mocopiの話と「HMDレスのフルトラ」について考える」です！お楽しみに！