0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

BlueskyをPythonでこねくり回した話

Posted at

これは「AprilKnights Advent Calendar 2024」の16日目の記事です。
https://adventar.org/calendars/10571

最近やたらユーザの増えてきたBluesky、先日2400万を超えたとか。
Threadsは3億見えてきたそうですが

APIも自由に利用できるため、我々エンジニアの玩具としても遊びがいのある環境です。
ということで、ちょっとこねくり回してみました。

具体的な実装方法だけ知りたい方は「オチ」までスキップ推奨

APIを叩いてみる

BlueskyのAPIについて

Blueskyには用途によっていくつかホストがあります。
その中で今回使ったものを紹介。

bsky.social

ポストの投稿などログインが必要なAPIはこちらで実行します。

public.api.bsky.app

ユーザ情報などログインしなくても参照できるものはこちらが利用できるようです。
パッと思いつく用途としてはTwitterでも時々見かけた「ブログに自身の呟きを埋め込む」とか?

Botを作るにあたって利用したAPI

ひとつの記事にまとめるには多すぎるので
実際使ったものを抜粋して紹介

  1. セッションを作成
    • com.atproto.server.createSession
  2. 投稿
    • com.atproto.repo.createRecord

最低限だとこれだけです。

あとは継続的に回すならこの辺でセッション管理して

  • com.atproto.server.refreshSession
  • com.atproto.server.getSession

画像をアップロードしたい場合は以下のAPIでアップロード、帰ってきたURLを投稿に含めればOK

  • com.atproto.repo.uploadBlob

ポストする際のJSONの構造についてはこの辺参照

Pythonでの実装例

import json
import requests
from datetime import datetime

post_text = 'hello bluesky!!'
user_name = '自身のアカウント名'
password = 'アプリパスワード' # Blueskyの「設定→高度な設定」から作成

# セッション作成
url = 'https://bsky.social/xrpc/com.atproto.server.createSession'
data = {'identifier': user_name, 'password': password}
headers = {'content-type': 'application/json'}
session = requests.post(url, data=json.dumps(data), headers=headers).json()

# 投稿
url = 'https://bsky.social/xrpc/com.atproto.repo.createRecord'
data = {
    'repo': session['did'],
    'collection': 'app.bsky.feed.post',
    'record': {
        'text': post_text,
        'createdAt': datetime.now().isoformat() + 'Z',
        'type': 'app.bsky.feed.post',
    }
}
headers = {
    'Authorization': 'Bearer ' + session['accessJwt'],
    'content-type': 'application/json'
}
requests.post(url, data=json.dumps(data), headers=headers)

APIを叩く際の注意点

https://docs.bsky.app/docs/advanced-guides/rate-limits
APIの実行にはレート制限があるのですが、どうも実際はこれより厳しいようです。
自動的に投稿するBotを運用していたところ、以下の状況で制限されるケースがありました。

あくまで本記事作成時点、状況から推測も含んだ話です。
将来厳しくなったり緩和されたりする可能性はあります。

レート制限に達するペースでAPI呼び出し

「レコード操作は1時間に5000ポイント、1日35000ポイントまで」というルールがありますが、どうも「これを超えるであろうペースで利用」すると制限されることがあるようです。

「1000件のレコードを作成」すると3000ポイント相当の消費になりますが、例えばこれを「1時間に1回、1分の間に1000件のレコードを作成」した場合、ルール上は問題ありませんが制限されるようです。
「1時間かけて1000件のレコードを作成」するよう調整しましょう。

レート制限ギリギリ

「createSessionは5分に30回、1日に300回まで」というルールに対して「5分ごとに1回の実行、1日288回、セーフ!ヨシ!」とかずっとやったら制限されました。:innocent:

「ギリギリセーフ」はNGっぽい、キチンとセッション管理しましょう。

最後に(オチ)

と、いくつかAPIと実装例を紹介しましたが、実はPythonで実装する場合には「atproto」のパッケージが存在しており、これを利用することでかなり簡単に実装をすることが可能です。

リファレンス斜め読みして実装する猪突猛進の阿呆はいるか! ここにいるぞ!

from atproto import Client

post_text = 'hello bluesky!!'
user_name = '自身のアカウント名'
password = 'アプリパスワード' # Blueskyの「設定→高度な設定」から作成

client = Client()
client.login(user_name, password)

post = client.send_post('hello bluesky!!')

尚、前述の通りセッションの作成はある程度制限があるようなので、ログイン後にセッション情報を出力&保存しておき、トークンの有効期限内ならセッションをそのまま使い回すと良さそうです。

# ログイン後にセッション情報を取得
client.login(user_name, password)
session_str = client.export_session_string()

# ---中略---

# 保持していたセッション情報で再ログイン
client.login(session_string=session_str)

send_imageを使えば画像付きポストも「アップロード」「投稿」の2工程で行う必要はない模様

with open('./image_1.jpg', 'rb') as f:
    img = f.read()
client.send_image(post_text, img, 'alt message')

最後の最後に

…と、ここまで下書きしてアドベントカレンダーの公開日まで記事を寝かせていたら、Raspberry Pi Zero 2Wにて「atprotoのpip installにコケる現象」が発生、APIを直接叩く形式であればコケようがないので、この記事が役立つ場面もある、かも。

ちなみに、pipのパッケージの依存関係が変なことになっていたようで、整理を兼ねて環境の再構築で解消。

0
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?