PythonでTwitter APIを利用するまで (Auth 2.0 with PKCE 認証)

はじめに

最近やりたいことが多くてツイッターする暇がないので、自分の代わりにツイッターやってくれるプログラムを組みたくて手を付けました。
将来的にはチャットBOTや情報収集AIのようなものも考えたいので、その分野に明るいPythonを選択。
認証部分はAuth 1.0aとAuth 2.0で選択ですが、やるなら最新であり他のサービスでも考え方が流用できるAuth 2.0を選択。

公式docが分かりにくいうえにPythonの公式サンプルがない部分もあり、また有志のPythonサンプルは昔のコード(Auth 1.0a利用)のものしか見つからなかったので、標準モジュールの中身を見たりテストプログラム走らせたりしながら動くものを作りました。
Auth 2.0利用のサンプルコードはいくばくか需要があるのではないかと思い、折角ですし作ったコードについて記事を書いてみます。

開発環境

• Windows 11 home
• Anaconda 3-2022.05 (Jupyter Notebook 6.4.8)
• Python 3.9.14
  ※ 認証確認ソースコードの方はPython 3.10.6(非Anaconda)でバッチ実行して正常終了を確認

必要なキー情報

Twitter Developersページで開発者アカウントを作り、Projects及びAppsを登録する必要があります。
そのうえで以下の情報が必要です。

• OAuth 2.0 Client ID and Client Secret 取得
  Keys and tokensタブから生成できます。同じページで取得できるConsumer Keys, Authentication Tokens [Bearer Token, Access Token and Secret] は不要。
• Redirect URI 入力
  User authentication settingsページから入力。ツイッターアカウントの権限を取得する際に必要ですが、今回は自分のアカウントだけなのでどんなURLでもいいです。

(やり方は探せば有志の方の記録が見つかると思いますので、ここでは割愛します)

ソースコード・初期認証

Auth2-1.py

import base64
import hashlib
import os
import re
import json
import requests
from requests.auth import AuthBase, HTTPBasicAuth
from requests_oauthlib import OAuth2Session

# 認証準備
client_id = "【自分のクライアントID】"
client_secret = "【自分のクライアントシークレット】"
redirect_uri = "https://127.0.0.1:3000/cb"
scopes = ["tweet.read", "tweet.write", "users.read", \
        "offline.access", "list.read", "like.read", "like.write"] 
# 任意の文字列とその変換形を作成
code_verifier = base64.urlsafe_b64encode(os.urandom(30)).decode("utf-8")
code_verifier = re.sub("[^a-zA-Z0-9]+", "", code_verifier)
code_challenge = hashlib.sha256(code_verifier.encode("utf-8")).digest()
code_challenge = base64.urlsafe_b64encode(code_challenge).decode("utf-8")
code_challenge = code_challenge.replace("=", "")

oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scopes)

auth_url = "https://twitter.com/i/oauth2/authorize"
authorization_url, state = oauth.authorization_url(
    auth_url, code_challenge=code_challenge, code_challenge_method="S256"
)

authorization_response = input(
    "Paste in the full URL after you've authorized your App:\n"
)

token_url = "https://api.twitter.com/2/oauth2/token"
auth = HTTPBasicAuth(client_id, client_secret)

token = oauth.fetch_token(
    token_url=token_url,
    authorization_response=authorization_response,
    auth=auth,
    client_id=client_id,
    include_client_id=True,
    code_verifier=code_verifier,
)
print(token)

# 認証と操作対象設定（事前準備ゴール）
access = token["access_token"]

# リクエスト作成/疎通確認
params = {"user.fields": "created_at,description"}
headers = {
    "Authorization": "Bearer {}".format(access),
    "User-Agent": "auth_test",
}
url = "https://api.twitter.com/2/users/me"
response = requests.request("GET", url, params=params, headers=headers)
if response.status_code != 200:
    raise Exception(
        "Request returned an error: {} {}".format(response.status_code, response.text)
    )
print("finish!")

解説

箇条書きで簡単に。概ね公式サンプルを踏襲したコードです。
★事前準備として、普段のブラウザで自分のアカウントでログインしておくこと

• scopesで利用する権限を指定する。内容は何となく分かると思いますが詳細は公式doc参照。
• スクリプト実行すると、authorization_url等が表示されるとともにinputを求められる。
• authorization_url で出てきたURLにブラウザでアクセスすると、自分のアカウントにログインした状態で認証画面が出てくる。
• 認証許可ボタンを押すとredirect_uriにパラメータ付きで飛ばされる。
  (ちゃんとwebページを用意して受け取れるようにするべきだと思うけど、今回はパラメータさえ分かればいい。127.0.0.1は自分自身の端末を指す。cbは一応call backの略だけど意味はなく、アクセスしてもエラー画面が出てくる。)
• エラー画面表示時、ブラウザのアドレスバーに表示されるURLはhttps～全文inputに突っ込む。
• tokenで得られるのが今回の目的のトークン ここにアクセスするための情報が記載されている。
  公式のサンプルコードではaccess_tokenしか利用していないが、refresh_tokenも後々使うので全文出力し控えておくこと。ここで結構つまずいて時間食われた。
• その下で疎通確認。「2/user/me」はトークンでログインしたユーザの情報を得るためのWeb API。
  requestメソッドの中身を変えるといろいろできる。

ソースコード・認証確認

Auth2-2.py

import requests
import json
import datetime
import time
from requests.auth import AuthBase, HTTPBasicAuth
from requests_oauthlib import OAuth2Session

# 初期設定(認証等)
def setting():
    with open('【作業用ディレクトリ】\\config.json', 'r', encoding='UTF-8') as f:
        json_config = json.load(f)
    # 認証確認
    params = {"user.fields": "created_at,description"}
    headers = {
        "Authorization": "Bearer {}".format(json_config["access_token"]),
        "User-Agent": "auth",
    }
    url = "https://api.twitter.com/2/users/me"
    response = requests.request("GET", url, params=params, headers=headers)
    if response.status_code != 200:
        if response.status_code == 401:
            # トークン期限が切れているので再取得
            client_id = json_config["client_id"]
            client_secret = json_config["client_secret"]
            redirect_uri = json_config["redirect_uri"]
            scopes = json_config["scopes"]
            oauth = OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scopes)
            auth = HTTPBasicAuth(client_id, client_secret)
            refreshed_token = oauth.refresh_token(
                client_id=client_id,
                auth=auth,
                token_url="https://api.twitter.com/2/oauth2/token",
                refresh_token=json_config["refresh_token"]
            )
            json_config["access_token"] = refreshed_token["access_token"]
            json_config["refresh_token"] = refreshed_token["refresh_token"]
            with open('【作業用ディレクトリ】\\config.json', 'w', encoding='UTF-8') as f:
                json.dump(json_config, f, indent=4, ensure_ascii=False)
        else:
            # 例外処理
            raise Exception(response.status_code, response.text)
    return json_config

# 以下メイン処理等を記述

解説

こちらも箇条書きで。
前提として、Twitter APIのアクセストークンは2時間で有効期限が切れるため、上記処理のようなリフレッシュが必要です。

• 必要な情報はconfig.jsonに記載しておき、最初に読み取るようにしている。ここに初期認証で得られたaccess_token及びrefresh_tokenを記載している。
• 疎通確認を行い、認証できないという反応が返ってきたらトークンのリフレッシュを行う。
• requests_oauthlib の中にそれらしい関数があったので、公式のcurl文を参考にパラメータを突っ込んでる。

注意事項

こちらが公式docにcurl文しかなく、Pythonだとどうするんだ…となって困った部分です。一応これで動くようになりました。
トークンの生成・リフレッシュができてるのでパラメータに不足はないはずですが、十分な理解がない中コードを組んでるので、逆に不要なパラメータが含まれてる可能性はあります。
複数回実行しても問題なく動いてるので、とりあえずこのまま運用しようという考えで放置しています。
理解が進めば整理したいと思います。その際にはこの記事にも反映します。

おわりに

公式docのとおりURLやパラメータを指定してリクエストしてあげると自分のアカウントの操作ができます。Pythonコードがきちんと書ければタイムライン取得もツイート検索もできますし、ツイート投稿もリツイートもできるようになります。
私のアカウントは今のところ、推しのYoutube配信情報を自動取得してくれています。他に何させるかは考え中。
いいねもAPIは用意されてるんですが、利用規約に「自動的な方法によるいいね禁止」と書いてあったので現状やらないようにしています。いわゆる迷惑行為にならないように、利用規約は順守したいですね。

いろいろやりたいと思ってElevated access(月200万ツイートまで取得可能)申請も通したので、せっかくですしもっと色々やりたいです。AWS及びGoogle Cloud(両方クラウドサービス)も活用して、「私の代わりにツイッターしてくれるシステム」の構築を鋭意進めていますので、ひと段落したらそちらも記事にしてみたいと思います。

以上。