プログラミング

BlueSky API を使ってリプライに反応してみた(python)

pythonでBlueSkyのAPIを使ってリプに反応する処理を作ってみました

実際に見てみよう

実際にはこのように動きます。いい感じですね

https://bsky.app/profile/quizwiki.bsky.social

作る

今回はフォローしているかつ、メンション付きの投稿を検出して、それに対してクイズをリプライする機能を作りました。

ざっくりの流れ

  • 通知欄を取得
  • メンション付きの投稿を対象に処理
    • フォローしているかどうか確認
    • 「ポケモンシルエットクイズ」が投稿内容に含まれているか確認
    • リプ
  • 通知を既読

参考

公式のサンプル（ほとんどこのまま使える。フォローしているかどうかは自前）

https://github.com/MarshalX/atproto/blob/main/examples/process_notifications.py

かなり久々にドキュメントを見ながら作った
https://atproto.blue/en/latest/readme.html

実装

準備

プログラムの中でログインが必要です。

他のAPIなどではAPIキーなどを発行したりしますが、今回はもっと簡単です。

設定＞アプリパスワード＞アプリパスワードを追加

適当な名前を付ける

アプリパスワードをコピー　この後使います！！！

ここまで出来たら準備完了です

コーディング

まずはサンプルコードをコピーしてきてさっき作ったアプリパスワードとhandle(xxxx.bsky.social)を入力しましょう！

from time import sleep

from atproto import Client

# how often we should check for new notifications
FETCH_NOTIFICATIONS_DELAY_SEC = 3


def main() -> None:
    client = Client()
    client.login('my-handle', 'my-password') # ここにさっき作ったパスワードとhandle(xxxx.bsky.social)

    # fetch new notifications
    while True:
        # save the time in UTC when we fetch notifications
        last_seen_at = client.get_current_time_iso()

        response = client.app.bsky.notification.list_notifications()
        for notification in response.notifications:
            if not notification.is_read:
                print(f'Got new notification! Type: {notification.reason}; from: {notification.author.did}')
                # example: "Got new notification! Type: like; from: did:plc:hlorqa2iqfooopmyzvb4byaz"

        # mark notifications as processed (isRead=True)
        client.app.bsky.notification.update_seen({'seen_at': last_seen_at})
        print('Successfully process notification. Last seen at:', last_seen_at)

        sleep(FETCH_NOTIFICATIONS_DELAY_SEC)


if __name__ == '__main__':
    main()

簡単に解説します

client.app.bsky.notification.list_notifications()

通知のリスト（ベルのマークを押したときに出てくるあれ）を取得します

if not notification.is_read:

通知を読んだかどうかを表しています。同じ通知に対して何度も反応しないようにするためのものです

client.app.bsky.notification.update_seen({'seen_at': last_seen_at})

引数に指定したseen_at（2024-02-12T00:19:08.912187+00:00）までの通知をすべて既読にします
この時刻ですがUTCにしてください。内部的にはこのタイムゾーンで管理してるかもです。（日本時間にするとうまく動かない）

for notification in response.notifications:

取得した情報を見ていきます。

author.viewer.followed_byがNoneではない場合この人は私のことをフォローしています。

uri, cid を使うとこの投稿が特定できていいねをしたりできます。

display_nameはユーザー名（設定していなければ''です）

record.textは本文です。この中に「ポケモンシルエットクイズ」という文言がある場合にリプを返します。

notification = {
    "author": ProfileView(
        did='did:plc:prn7imjv3flekrh4k4z5mpof',
        handle='cpord.bsky.social',
        avatar=None,
        description=None,
        display_name='',
        indexed_at='2024-02-12T05:39:57.466Z',
        labels=[],
        viewer=ViewerState(
            blocked_by=False,
            blocking=None,
            blocking_by_list=None,
            followed_by='at://did:plc:prn7imjv3flekrh4k4z5mpof/app.bsky.graph.follow/3kl7psks5ig2x', 
            following=None,
            muted=False,
            muted_by_list=None,
            py_type='app.bsky.actor.defs#viewerState'
        ),
        py_type='app.bsky.actor.defs#profileView'
    ),
    "cid": 'bafyreicxyuwd5m7pbxd3kin3xddll4dkti4h3q4tyjqxj2lm3tlhq3ekgi',
    "indexed_at": '2024-02-13T23:47:53.344Z',
    "is_read": False,
    "reason": 'mention',
    "record": Record(
        created_at='2024-02-13T23:47:53.344Z',
        text='@quizwiki.bsky.social',
        embed=None,
        entities=None,
        facets=[
            Main(
                features=[
                    Mention(
                        did='did:plc:owqmnbrxfvl4ktlzkswjlxfr',
                        py_type='app.bsky.richtext.facet#mention'
                    )
                ],
                index=ByteSlice(
                    byte_end=21,
                    byte_start=0,
                    py_type='app.bsky.richtext.facet#byteSlice'
                ),
                py_type='app.bsky.richtext.facet'
            )
        ],
        labels=None,
        langs=['ja'],
        reply=None,
        tags=None,
        py_type='app.bsky.feed.post'
    ),
    "uri": 'at://did:plc:prn7imjv3flekrh4k4z5mpof/app.bsky.feed.post/3kldk4xicw22p',
    "labels": [],
    "reason_subject": None,
    "py_type": 'app.bsky.notification.listNotifications#notification'
}

変更点

リプを返すための情報とフォローしているかどうかの判定を行います

for notification in response.notifications:
            if not notification.is_read:  
                if notification.reason == "mention":
                  is_follower = notification.author.viewer.followed_by is not None
                  user_name = notification.author.display_name if notification.author.display_name else notification.author.handle
                  root_ref = models.create_strong_ref(notification)
                  if not is_follower:
                    print("[INFO]" + user_name + "にフォローするように依頼しました")

if notification.reason == "mention":

これによって通知がメンションであることを確認します。

‘like’, ‘repost’, ‘follow’, ‘mention’, ‘reply’, ‘quote’ があります

is_follower = notification.author.viewer.followed_by is not None

メンションしてきた人がフォローしているか確認します

user_name = notification.author.display_name if notification.author.display_name else notification.author.handle

名前を取得します（ログ用）display_nameが設定されていない場合はhandle(xxxx.bluesky.social)を取得

root_ref = models.create_strong_ref(notification)

結構大事。

これを使ってリプライをします。これを指定してあげないとリプになりません。

client.send_post(
    text="問題を受け取るにはフォローをお願いします！",
    reply_to=models.AppBskyFeedPost.ReplyRef(parent=root_ref, root=root_ref),
)

いったんおしまい

こんなところでしょうかサンプルコードとドキュメントを見ながら開発するのは

新しいものをやっている感じがして楽しいですね。

Twitterのような魔境とは違う環境。かなりいいね。

反響があればもっと詳しく書きますのでコメントとかください！

リンクカードの投稿方法とかもまた教えますね

-プログラミング
-