📝

Blueskyのモデレーションリストにユーザーを追加する（Python）

2024/05/13に公開

Python

更新履歴

2024/05/14 完全一致検索の調査を追加

はじめに

APIで投稿するサンプルはいくつか見つかるものの、検索やモデレーションリストの追加等のサンプルが無かったので作りました。
実際に使う場合は、後述の注意点等を読んで適宜修正してください。

事前準備

Blueskyでの作業

サンプルプログラムの書き換えに必要な情報の取得と、モデレーションリストの作成を行います。

プロフィール画面等から自分のID（例：example.bsky.social）を取得する
アプリパスワードを発行し、取得する
- 「設定」－「アプリパスワード」画面を開き、「アプリパスワードを追加」ボタンから作成
- 参考：https://docs.tokimeki.blue/ja/apppassword
Blueskyでモデレーションリストを作成する
- 「設定」－「モデレーション」－「モデレーションリスト」画面を開き、「＋新規」ボタンから作成
作成したモデレーションリスト画面を開き、URL（の末尾）を取得する

サンプルプログラムの書き換え

必須設定

login_id、app_pass、uri_endの値を書き換えてください。

login_id = "自分のID"  # 例：example.bsky.social
app_pass = "アプリパスワード"  # BlueSkyの「設定」－「アプリパスワード」画面から作成したものを転記する

# 対象のモデレーションリストをブラウザで開き、URLの末尾を転記する
# 例：下記の様なアドレスの場合 "1234567890abc" を転記する
#    https://bsky.app/profile/example.bsky.social/lists/1234567890abc
uri_end = "モデレーションリストの末尾"

必要なら都度修正

サンプルプログラムの動作を見たいだけなら、書き換えなくても動作します。
実際にミュート・ブロックしたいユーザーを追加する場合は、最低でもsearch_wordを書き換えてください。
検索文字列については、後述の検索文字列（クエリ文字列）についても参照してください。

# ユーザー検索の文字列
# ※ユーザー名、プロフィールに対して検索されます
# ※日本語の場合、Bluesky側の単語分解の仕方によって、予期せぬユーザーが含まれる可能性があります
search_word = '"テスト用"'  # ダブルクォーテーション付きの文字列の方が比較的厳密

limit = 25  # ユーザー検索時に一度に取得するユーザー数（1～100、デフォルト値は25）
is_all_get = False  # 対象ユーザーを全件取得するかどうか（テスト時はFalse、本番時はTrueを推奨）

# 初期値はブランク、前回の続きを取得するときは後続処理で取得したcursorを設定する
cursor = ""

コマンドプロンプトで pip を実行

pip install requests

サンプルプログラムの流れ

認証用のJWTとDIDを取得する
com.atproto.server.createSession 関数を呼ぶ
モデレーションリストに追加するユーザーの情報を取得する
app.bsky.actor.searchActors 関数を呼ぶ
一度に取得可能なユーザー数は最大100件
cursorを指定することで、続きを取得可能
検索文字列にヒットしたユーザーをすべて登録する場合はis_all_getフラグをTrueに変更してから実行する
モデレーションリストにユーザーを追加する
com.atproto.repo.createRecord 関数を呼ぶ
一括登録の関数はなさそうなので、1件ずつループする

サンプルプログラム

長いので折りたたんでます

import sys
import time
import requests
from datetime import datetime, timezone
from dataclasses import dataclass


# ユーザー情報をまとめたデータクラス
@dataclass
class UserInfo:
    display_name: str
    did: str


# モデレーションリストにユーザーを追加する
def main():
    # ※※※ 実行前に以下の login_id ～ is_all_get を書き換えてください ※※※

    login_id = "自分のID"  # 例：example.bsky.social
    app_pass = "アプリパスワード"  # BlueSkyの「設定」－「アプリパスワード」画面から作成したものを転記する

    # 対象のモデレーションリストをブラウザで開き、URLの末尾を転記する
    # 例：下記の様なアドレスの場合 "1234567890abc" を転記する
    #    https://bsky.app/profile/example.bsky.social/lists/1234567890abc
    uri_end = "モデレーションリストの末尾"

    # ユーザー検索の文字列
    # ※ユーザー名、プロフィールに対して検索されます
    # ※日本語の場合、Bluesky側の単語分解の仕方によって、予期せぬユーザーが含まれる可能性があります
    search_word = '"テスト用"'  # ダブルクォーテーション付きの文字列の方が比較的厳密

    limit = 25  # ユーザー検索時に一度に取得するユーザー数（1～100、デフォルト値は25）
    is_all_get = False  # 対象ユーザーを全件取得するかどうか（テスト時はFalse、本番時はTrueを推奨）

    # 初期値はブランク、前回の続きを取得するときは後続処理で取得したcursorを設定する
    cursor = ""

    # 認証用のJWTとDIDを取得
    access_jwt, my_did = get_jwt_and_did(login_id, app_pass)

    # モデレーションリストに追加するユーザーの情報を取得する
    users, next_cursor = get_users_info(access_jwt, search_word, limit, cursor, is_all_get)
    print(f"検索文字列: {search_word} の検索結果は{len(users)}件です")

    # 対象のモデレーションリストのURI
    list_uri = f"at://{my_did}/app.bsky.graph.list/{uri_end}"

    # モデレーションリストにユーザーを登録する
    add_users(my_did, access_jwt, list_uri, users)

    # 全件取得しない場合、且つ続きがある場合は次回のためにCursorを出力
    if next_cursor:
        print(f'次回、この続きを検索する場合はcursorに"{next_cursor}"を設定してください')


# 認証用のJWTとDIDを取得
def get_jwt_and_did(login_id, app_pass):
    print("---認証用のJWTとDIDを取得---")

    # 参考：https://docs.bsky.app/docs/api/com-atproto-server-create-session
    url = "https://bsky.social/xrpc/com.atproto.server.createSession"
    body = {"identifier": login_id, "password": app_pass}
    headers = {"Content-Type": "application/json; charset=UTF-8"}

    response = requests.post(url, headers=headers, json=body)
    print(f"StatusCode: {response.status_code} {response.reason}")

    return response.json()["accessJwt"], response.json()["did"]


# モデレーションリストに追加するユーザーの情報を取得する
def get_users_info(access_jwt, search_word, limit, cursor, is_all_get):
    print("---ユーザー情報を取得---")
    users = []

    # 参考：https://docs.bsky.app/docs/api/app-bsky-actor-search-actors
    url = f"https://bsky.social/xrpc/app.bsky.actor.searchActors"
    params = {
        "q": search_word,
        "limit": limit,
        "cursor": cursor,
    }
    headers = {
        "Accept": "application/json",
        "Authorization": f"Bearer {access_jwt}",
        "Content-Type": "application/json; charset=UTF-8",
    }

    while True:
        response = requests.get(url, headers=headers, params=params)
        print(f"Cursor: {params['cursor']} StatusCode: {response.status_code} {response.reason}")

        # モデレーションリストに追加するユーザーの情報を取得
        for actors in response.json()["actors"]:
            users.append(
                UserInfo(
                    # display_name（ユーザー名）は後続処理のログ出力用に取得する
                    display_name=actors.get("displayName", ""),
                    did=actors["did"],
                )
            )

        # 最後まで検索した場合、responseにはcursorキーが含まれないので処理を終了する
        if not is_all_get or not "cursor" in response.json():
            break
        else:
            # 続きを取得するため、パラメータのcursorを更新する
            params["cursor"] = response.json()["cursor"]

    return users, response.json().get("cursor", "")


# モデレーションリストにユーザーを登録する
def add_users(my_did, access_jwt, list_uri, users: list[UserInfo]):
    print("---モデレーションリストにユーザーを追加---")

    # 参考：https://docs.bsky.app/docs/api/com-atproto-repo-create-record
    # 参考：https://docs.bsky.app/docs/tutorials/user-lists#add-a-user-to-a-list
    url = f"https://bsky.social/xrpc/com.atproto.repo.createRecord"

    for user in users:
        print(f"対象ユーザー DisplayName: {user.display_name} DID: {user.did}")

        body = {
            "repo": my_did,
            "collection": "app.bsky.graph.listitem",
            "record": {
                "type": "app.bsky.graph.listitem",
                "subject": user.did,
                "list": list_uri,
                "createdAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
            },
        }

        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {access_jwt}",
            "Content-Type": "application/json; charset=UTF-8",
        }

        response = requests.post(url, headers=headers, json=body)
        print(f"StatusCode: {response.status_code} {response.reason}")

        # 負荷防止のため、一応0.2秒待つ
        time.sleep(0.2)


if __name__ == "__main__":
    main()
    sys.exit(0)

プログラム実行後の作業

Blueskyでモデレーションリストをミュート、またはブロック用に登録する
- リストを作っただけだとミュート、ブロックの対象にはなりません
  - サンプルプログラムの動作を見るだけなら、ミュート、ブロックの登録は不要です
- ミュート、ブロックの登録・解除は任意のタイミングで可能です

サンプルプログラムの注意点

実際に活用する場合は適宜修正してください。

サンプルなので最低限の処理しか書いてません
- 必要に応じてエラー処理やログ出力等を追加してください
- サンプルプログラムではDIDとDisplayName（ユーザー名）しか取得してませんが、searchActorsで取得したresponse.json()をファイル出力するか、必要な情報を抽出して保存すると確認しやすいです
  - descriptionキーにはプロフィールの文字列が格納されています
モデレーションリストはミュート・ブロックを目的としているため、ユーザーの誤追加に気をつけてください
- 検索だけの処理をするプログラムを作るか、サンプルプログラムのadd_users関数を呼ぶ箇所をコメントアウトし、問題なさそうならユーザー追加することをお勧めします
モデレーションリストに追加済みのユーザーを再度追加しても正常終了します
- 定期実行したり大量にユーザーを追加する場合は、必要以上に負荷を掛けないようご注意ください
  - 個人的には、ローカルDBに追加済みのユーザー情報と前回実行時のCursorを保存して、事前の重複チェックや前回のCursorを利用して続きの検索等を実装しています
post時のcreatedAtはdatetime.now().isoformat()だとエラーになりました

エラーメッセージ：createdAt must be an valid atproto datetime (both RFC-3339 and ISO-8601)
- 下記サイトに記述の通りに記載すれば問題ありません
  https://docs.bsky.app/docs/advanced-guides/posts#post-record-structure
```
# Fetch the current time
# Using a trailing "Z" is preferred over the "+00:00" format
now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
```

検索文字列（クエリ文字列）について

Blueskyは日本語の形態素解析（単語分解）にはKuromojiを使っているらしいです
検索文字列にはLuceneのクエリ構文が推奨されています
- https://docs.bsky.app/docs/api/app-bsky-actor-search-actors
  
  Search query string. Syntax, phrase, boolean, and faceting is unspecified, but Lucene query syntax is recommended.
単語分解の結果が検索に影響するため、（日本語の場合は特に）ダブルクォーテーションで括っても完全一致にならないケースがあります
- 例：「フリーエンジニア」という単語の場合
  - search_word = '"フリーエンジニア"'
    - 5件ヒット
    - 文字列としてダブルクォーテーションを追加して完全一致を求めても、"フリーエンジニア"、"フリーのエンジニア"、"フリーダムエンジニア" 等がヒットしました
  - search_word = '"フリー" && "エンジニア"'
    - 60件ヒット
    - AND条件で検索したのでヒット数が増加しました
  - search_word = "フリーエンジニア"
    - 60件ヒット
    - AND条件と同じ結果でした
  - search_word = '"フリー" || "エンジニア"'
    - 7562件ヒット
    - OR条件で検索したのでヒット数が一気に増加しました
  - search_word = '"フリー && エンジニア"'
    - 3件ヒット
    - "フリーエンジニア"の単語と完全一致したユーザーがヒットしました
  - search_word = '"フリーエンジニア"' #間にスペース
    - 3件ヒット
    - '"フリー && エンジニア"'と同じ結果でした

完全一致検索の調査

結論：ダブルクォーテーションで括り、その中でAND検索をした場合はおそらく文字列単位の完全一致検索が可能
※AND検索と言いつつ、実際には半角スペースで区切っていれば良く、ANDやORの記号は関係なさそうでした

例：search_word = '"フリー && エンジニア"'（または'"フリーエンジニア"'）で検索した場合

displayName（ユーザー名）またはdescription（プロフィール）の中に両方の単語が続けて入っていればヒットする
2単語の間にスペース、記号が含まれていてもヒットする
- 例：フリー +-*/エンジニア
- すべての記号かどうかは未確認
2単語の間に別の単語が含まれているとヒットしない
- 例：フリーのエンジニア
displayNameに"フリー"、descriptionに"エンジニア"だとヒットしない
- 'フリー && エンジニア' のようにダブルクォーテーションを付けない場合はヒットする
長文や複雑な単語の場合、単語分解の結果によっては正常に動作しない可能性がある（未検証）
やってることは同じなので、公式の検索機能でも"フリー && エンジニア"と入力すれば良い

実際の検索結果（長いので折りたたんでます）

自分のプロフィール欄に文字列を記入して、検索にヒットするかどうかを調査。
自分だけがヒットする「暇読む」という単語をベースに、条件を変えて検索を実行。
ヒットした場合は、「結果」に「○」を記載。

暇読む（ベースの文字列）

検索文字列結果

'"暇読む"' ○

'"暇読む"' ○

'"暇 && 読む"' ○

'"暇 || 読む"' ○

検索文字列	結果
'"暇読む"'	○
'"暇読む"'	○
'"暇 && 読む"'	○
'"暇 \|\| 読む"'	○

読む暇（単語の並び順変更）

検索文字列	結果
'"読む暇"'	○
'"読む暇"'	○
'"読む && 暇"'	○
'"読む \|\| 暇"'	○
'"暇読む"'	×
'"暇読む"'	×
'"暇 && 読む"'	×
'"暇 \|\| 読む"'	×

暇を読む（間に一文字追加）

検索文字列結果

'"暇を読む"' ○

'"暇読む"' ×

'"暇読む"' ×

'"暇 && 読む"' ×

'"暇 || 読む"' ×
暇読む（間に半角スペースを2つ追加）

検索文字列結果

'"暇読む"' ○

'"暇読む"' ○

'"暇 && 読む"' ○

'"暇 || 読む"' ○
暇　読む（間に全角スペースを1つ追加）

検索文字列結果

'"暇読む"' ○

'"暇読む"' ○

'"暇 && 読む"' ○

'"暇 || 読む"' ○

'"暇　読む"'（全角スペース） ○
暇 #読む（間にスペース＋#を追加してハッシュタグにする）

検索文字列結果

'"暇読む"' ○

'"暇読む"' ○

'"暇 && 読む"' ○

'"暇 || 読む"' ○
暇#読む（間にスペースなしで#を追加し、ハッシュタグとしては無効化）

検索文字列結果

'"暇読む"' ○

'"暇読む"' ○

'"暇 && 読む"' ○

'"暇 || 読む"' ○
暇+-*/読む（間に記号4種を追加）

検索文字列結果

'"暇読む"' ○

'"暇読む"' ○

'"暇 && 読む"' ○

'"暇 || 読む"' ○
暇読む+-*/（間にスペース＋末尾に記号4種を追加）

検索文字列結果

'"暇読む"' ○

'"暇読む"' ○

'"暇 && 読む"' ○

'"暇 || 読む"' ○
ユーザー名に「暇」プロフィールに「読む」

検索文字列結果

'"暇読む"' ×

'"暇読む"' ×

'"暇 && 読む"' ×

'"暇 || 読む"' ×

'暇 && 読む'（ダブルクォーテーションを付けない） ○

検索文字列	結果
'"暇を読む"'	○
'"暇読む"'	×
'"暇読む"'	×
'"暇 && 読む"'	×
'"暇 \|\| 読む"'	×

検索文字列	結果
'"暇読む"'	○
'"暇読む"'	○
'"暇 && 読む"'	○
'"暇 \|\| 読む"'	○

検索文字列	結果
'"暇読む"'	○
'"暇読む"'	○
'"暇 && 読む"'	○
'"暇 \|\| 読む"'	○
'"暇　読む"'（全角スペース）	○

検索文字列	結果
'"暇読む"'	○
'"暇読む"'	○
'"暇 && 読む"'	○
'"暇 \|\| 読む"'	○

検索文字列	結果
'"暇読む"'	○
'"暇読む"'	○
'"暇 && 読む"'	○
'"暇 \|\| 読む"'	○

検索文字列	結果
'"暇読む"'	○
'"暇読む"'	○
'"暇 && 読む"'	○
'"暇 \|\| 読む"'	○

検索文字列	結果
'"暇読む"'	○
'"暇読む"'	○
'"暇 && 読む"'	○
'"暇 \|\| 読む"'	○

検索文字列	結果
'"暇読む"'	×
'"暇読む"'	×
'"暇 && 読む"'	×
'"暇 \|\| 読む"'	×
'暇 && 読む'（ダブルクォーテーションを付けない）	○

バグっぽい挙動について

searchActorsの検索結果でresponseのactorsは空のリスト[]なのに、Cursorが25を返すケースが発生しました（25件正常取得したと認識されてるのにユーザー情報が取得出来てない）
- 検索文字列を別の値に変更したら正常に取得出来ました
- 元の文字列に戻すと同様の事象が発生しました
- 別の検索処理を行うPythonファイルでは同じ単語でも正常に動作しました
1回しか発生してないので微妙なところですが、同様の事象が発生した場合は別ファイルとして実行すると改善する可能性があります

モデレーションリストの注意点

少なくとも現時点（2024/05/13時点）では下記の状況です。
モデレーションリストは便利な機能ですが、安易に他者のリストを使うのも考え物です。

リストを非公開に出来ません
自分が作ったリストを他者が利用することが可能です
他者が作ったリストを検索する機能がありません（多分）
他者が自分のリストを利用しても気づけません
他者が作ったモデレーションリストは、中身が変わっても検知できません
- 特に登録ユーザー数が多いリストは、中身を確認するのも難しいです
リストの変更をしても画面上の反映に時間が掛かるときがあります
- リストを削除したり名前を変えても、しばらく反映されなかった等

公式ドキュメント

https://docs.bsky.app/docs/get-started
- ページによってはTypeScriptしか記述が無かったりとまだまだ微妙な感じですが、Tutorialsは眺めておくと良いです
- 今回はatprotoを使っていないため、主に参照したのはHTTP Referenceのページです