Blueskyの特定のフィードの投稿のみをJSONファイルに抽出するプログラムを作った

1.プログラム

GetFeedPost.py

import requests
import json

# ログイン処理
url = "https://bsky.social/xrpc/com.atproto.server.createSession"

data = {
    "identifier": "example.bsky.social",
    "password": "example12345hoge"
}

headers = {
    "Content-Type": "application/json; charset=UTF-8"
}

response = requests.post(url, data=json.dumps(data), headers=headers)
print(response)

# このaccessJwtを後の処理で使う
accessJwt = response.json()["accessJwt"]

#フィード取得
url = "https://bsky.social/xrpc/app.bsky.feed.getFeed/"
params = {
    "feed": "at://did:plc:q6gjnaw2blty4crticxkmujt/app.bsky.feed.generator/cl-japanese",
    "limit": 1
}

headers = {
    "Authorization": f"Bearer {accessJwt}"
}

response = requests.get(url, params=params, headers=headers)
responseJSON = response.json()

# レスポンスデータからポスト内容と作成日を抽出する
post_data = {
    "id": 0,
    "text": responseJSON["feed"][0]["post"]["record"]["text"],
    "createdAt": responseJSON["feed"][0]["post"]["record"]["createdAt"]
}

# JSONファイル名
json_file = 'post_data.json'

# 既存の JSON ファイルを読み込む
try:
    with open(json_file, 'r', encoding='utf-8') as file:
        existing_data = json.load(file)
except FileNotFoundError:
    existing_data = []

if existing_data:
    last_data = existing_data[-1]
    if last_data.get("text") == post_data["text"] and last_data.get("createdAt") == post_data["createdAt"]:
        print("最新の情報はすでに追加されています。処理を終了します。")
        exit()

    last_id = existing_data[-1]["id"]
    post_data["id"] = last_id + 1

existing_data.append(post_data)

# 抽出したデータをJSONファイルに書き込む
with open(json_file, 'w', encoding='utf-8') as outfile:
    json.dump(existing_data, outfile, indent=2, ensure_ascii=False)

print(f"ポストデータの抽出が完了しました。データは '{json_file}' に追記されました")

2.解説

このプログラムは、実行されるたびJapanese Clusterというフィードの最新の投稿をpost_data.jsonに保存します。
ついでに順にidを振っています。
出力例：

post_data.json

[
  {
    "id": 0,
    "text": "カレーライス食べたと思ったら空を飛んでいた",
    "createdAt": "2024-01-01T12:41:32.848Z"
  },
  {
    "id": 1,
    "text": "本日の昼食はこれ......\nカツカレーうまい！",
    "createdAt": "2024-01-01T12:41:38.694Z"
  },
  {
    "id": 2,
    "text": "カレーライスは辛いのが好き",
    "createdAt": "2024-01-01T12:41:43.074Z"
  }
]

前半部分では、ログインしaccessJwt(アクセストークン的なやつ)を取得しています。これによりAPIがデータを取得できるようになります。
（一度取得しても時間がたつと使えなくなるので、実行するたびにログインしなおすようにしています。）

一度に何回も実行する場合はログイン部のみ先に実行し、取得したaccessJwtを使用して投稿を取得するようにすると、APIのレート制限にかかりにくくなります。
(私の環境では大体30分ぐらいでaccessJwtが使用できなくなりました。)

後半部分では、指定したフィードの最新の投稿を取得し、その前に保存された内容と比較しています。同じ場合は処理を終了し、違う場合は追記して保存するようにしています。

3.余計な話

特定のフィードの最新の投稿を取得するなんていうのは誰もやっておらず、先人様達が作った別のコードやATプロトコルの仕様書を見比べつつやってみました。
パラメータについての説明がどこにもないせいで、これだけに二日かかりました。

フィード取得のところ

#フィード取得
url = "https://bsky.social/xrpc/app.bsky.feed.getFeed/"
params = {
    #なんでurlとfeedで書いてあることが違うのに動くんだよ！？
    "feed": "at://did:plc:q6gjnaw2blty4crticxkmujt/app.bsky.feed.generator/cl-japanese",
    "limit": 1
}

正直なんで動いてるのか自分でもわかっていないです()

4.参考にさせていただいたサイト

↓貴重な日本語記事様......!これ全部読めば大体APIの使い方がわかるぞ！

↓私が記事内で「仕様書」と呼んでいたもの。情報は広く浅くって感じなので、ほかの情報と併せて読まないとよくわからないです。

↓パラメータはこちらを参考に書きました。このサイトがなければ完全に詰んでました。

↓「at://」から始まる AT URI と呼ばれるルールのようです。何言ってるか分からないですが、例があるので真似して書きましょう()