こんにちは～ NRLH LAB の なぎ です。
今日は、
「自分の X のアカウントからブックマーク取得をしてみたよ～」
というのを開発日記として残したいと思います。
経緯
今、うちのAIアシスタントに新しい機能を実装していて、その中でXのアカウントの情報が欲しかったので実装をしてみました。
↓うちのAIアシスタントについて
XのAPIに触るのは始めてなので、色々調べながら...
方法
言語: Python
ライブラリ: tweepy
API: X API V2 Free
実装
自分のアカウントから 「ブックマーク」 を取得します。
以下を参考にしました。
Authentication — tweepy 4.14.0 documentation
コードを表示する
from logging import getLogger import webbrowser import json import requests import tweepy import config ResponseType = requests.Response # 「ブックマーク」データクラス class Bookmark: def __init__(self, id: str, text: str): self._id = id self._text = text @property def id(self): return self._id @property def text(self): return self._text def __str__(self): return f"'id': '{self.id}', `text`: '{self.text}'" # X API の認証のスコープ # https://docs.x.com/fundamentals/authentication/oauth-2-0/authorization-code SCOPES = ["tweet.read", "users.read", "offline.access", "bookmark.read"] logger = getLogger(__name__) def get_bearer_tokens(is_PKCE = True) -> dict[str, str]: if is_PKCE: # OAuth 2.0 Flow with PKCE による認証 oauth2_user_handler = tweepy.OAuth2UserHandler( client_id=config.X_OAUTH2_CLIENT_ID, redirect_uri="https://x.com/", scope=SCOPES, client_secret=config.X_OAUTH2_CLIENT_SECRET ) authorization_url = oauth2_user_handler.get_authorization_url() logger.info(f"認証フロー（OAuth 2.0 Flow with PKCE）のため、認証URLをブラウザで起動します。: {authorization_url}") webbrowser.open(authorization_url) logger.info("【要入力】アプリにアクセスを許可した後、リダイレクトされたURLをここに入力してください。") response_url = input() token = oauth2_user_handler.fetch_token(response_url) return {"bearer_token": token["access_token"]} else: return {"bearer_token": config.X_OAUTH_BEARER_TOKEN} def make_client(is_PKCE = True): """ X API を呼び出すためのクライアント（tweepy.Client）を作成する Parameters ---------- is_PKCE: bool 「OAuth 2.0 Flow with PKCE」での認証フローでクライアントを作成するかどうか。 Returns ------- client: tweepy.Client 作成したクライアント（tweepy.Client）のインスタンス """ bearer_tokens = get_bearer_tokens(is_PKCE) return tweepy.Client( bearer_token=bearer_tokens["bearer_token"] , consumer_key=config.X_API_KEY , consumer_secret=config.X_API_KEY_SECRET , access_token=config.X_OAUTH_ACCESS_TOKEN , access_token_secret=config.X_OAUTH_ACCESS_TOKEN_SECRET , return_type=ResponseType ) def get_bookmarks_response(client: tweepy.Client): CACHE_FILEPATH = ".\\aia300\\data\\x\\cache\\get_bookmarks.json" response = None try: # 公式では max_results <= 800 と書いてあるが、400 Bad Requestエラー # 1 ~ 100 とレスポンスで返ってきた（Free版だから？） response = client.get_bookmarks(max_results=100).json() # キャッシュ（json形式でファイルに保存） with open(CACHE_FILEPATH, 'w', encoding='utf-8') as file: json.dump(response, file, indent=2, ensure_ascii=False) except tweepy.TooManyRequests: # リクエスト数overなので、キャッシュから取得 logger.warning("リクエスト数overのため、過去に取得した最新のデータを参照します。") with open(CACHE_FILEPATH, 'r', encoding='utf-8') as file: response = json.load(file) return response["data"] def get_bookmarks(client: tweepy.Client) -> list[Bookmark]: """ ブックマーク取得（最大: 最新100個） Parameters ---------- client: tweepy.Client クライアント Returns ------- bookmarks: list[Bookmark] 取得したブックマークのリスト Notes ----- X API Free をしているので、リクエスト上限は "1 request / 15 min" です。 https://developer.x.com/en/portal/products 上限に達していた場合は、過去に取得した最新データ（ローカル保存）を返します。 """ response = get_bookmarks_response(client) bookmarks = [] for item in response: bookmarks.append(Bookmark(id=item["id"], text=item["text"])) return bookmarks # X API 呼び出し処理を実装したクラス class X: def __init__(self): self._client = make_client() @property def client(self): return self._client def get_bookmarks(self): return get_bookmarks(self.client)
取得したデータ（一部）
{ "data": [ { "text": "今回の記事（https://t.co/50qVWkXhMU）はこのRPのプロンプトを使わせていただき、校正しました。これとてもおすすめです！ https://t.co/Z9NBp7x1z6", "id": "2000914967563002107", "edit_history_tweet_ids": [ "2000914967563002107" ] }, { "text": "独学でモデリングを覚えた方にぜひ読んでほしい、\n ゲーム業界のキャラモデルが“実際どんな仕様で作られてるか” をまとめた記事を書きました。\n今回はスマホ向けのローポリ寄りキャラを例に、\n仕様・データ管理・工程・スケジュール感まで具体的に紹介しています。\n\nあくまで一例なので、", "id": "1994053792317501859", "edit_history_tweet_ids": [ "1994053792317501859" ] }, { "text": "【Unity】汎用的なジャンプ機構の実装例 - コヨーテタイム・先行入力・多段ジャンプ・可変ジャンプを共通化する | watabe_h\nhttps://t.co/qOkdBkMj8l", "id": "1996234273922125882", "edit_history_tweet_ids": [ "1996234273922125882" ] }, { "text": "【インタビュー】サイバーエージェントのゲーム事業部には「プロジェクト付け」でも「研究開発」でもない、グラフィックやシステム含めた“技術課題拾いまくりエンジニア組織”があるそうだ。なぜ？なんのために？求人中らしいので話を聞いた [AD]\nhttps://t.co/lwL1mxrZVw https://t.co/FsNLg2bQDt", "id": "2000400943247142959", "edit_history_tweet_ids": [ "2000400943247142959" ] }, { "text": "言語化が下手な人は5つのタイプに分けられる｜すてぃお @suthio_ https://t.co/xRW7WqkUG4", "id": "2000723816125988910", "edit_history_tweet_ids": [ "2000723816125988910" ] }, { "text": "【『原神』級のゲームはこうして生まれた──PCオンライン黎明期から原神・Black Myth誕生までの20年と日本の勝ち筋】\n 中国ゲームがなぜ高品質を量産できるのか。20年の歴史を紐解き、日本メーカーの現実的な勝ち筋を考察してみた\nhttps://t.co/Tr7JtJxJP5", "id": "1999677432455119204", "edit_history_tweet_ids": [ "1999677432455119204" ] } ], "meta": { "result_count": 6 } }
ポイント
認証
ブックマーク取得には、「OAuth 2.0 Flow with PKCE」 を使用して認証していないといけないようでした。
tweepyの リファレンス を読みその通りにコードを書けばよいのですが...
access_token = oauth2_user_handler.fetch_token(
"Authorization Response URL here"
)
client = tweepy.Client("Access Token here")
この部分、tweepy.Client()は、
def __init__(self , bearer_token=None , consumer_key=None , consumer_secret=None , access_token=None , access_token_secret=None , *, return_type=Response, wait_on_rate_limit=False ):
と定義されていますが、bearer_token=にaccess_tokenのデータをセットするのが正しいです。（ややこしい）
def __init__(self , bearer_token=None # 〇ここに oauth2_user_handler.fetch_token()["access_token"] をセット , consumer_key=None , consumer_secret=None , access_token=None # ×ここではない , access_token_secret=None , *, return_type=Response, wait_on_rate_limit=False ):
access_token=の方に渡してしまっていてここでかなり詰まりました...
キャッシュ
X の API はプランによってリクエストが制限されています。
Freeプラン（無料プラン）では一部APIが利用できなかったり、リクエスト数の制限が厳しいです。
今回の「ブックマーク取得」は 1リクエスト / 15分 / 1ユーザー となっています。
データの取得に失敗しても1リクエスト消費扱いになるので、
レスポンスはキャッシュ（ファイルに保存）し、制限によってデータが取得できなかった場合はそのファイルから読み込む
という形にしました。（今回の機能ではそこまでリアルタイムでなくてもよいので）
def get_bookmarks_response(client: tweepy.Client): CACHE_FILEPATH = ".\\aia300\\data\\x\\cache\\get_bookmarks.json" response = None try: response = client.get_bookmarks(max_results=100).json() # キャッシュ（json形式でファイルに保存） with open(CACHE_FILEPATH, 'w', encoding='utf-8') as file: json.dump(response, file, indent=2, ensure_ascii=False) except tweepy.TooManyRequests: # リクエスト数overなので、キャッシュから取得 logger.warning("リクエスト数overのため、過去に取得した最新のデータを参照します。") with open(CACHE_FILEPATH, 'r', encoding='utf-8') as file: response = json.load(file) return response["data"]
おわりに
ということで、
「Xからブックマークを取得してみたよ～」
という日記でした。
XのAPI君、もうちょっと仲良くしてほしいな～ﾁﾗｯ
実際に今開発している機能にどう組み込んだかはまた後日日記にしたいと思います。
ではまた。
