Twitter APIを使って自動でTweetする #Python

AWSのLambdaなどサーバレス環境とTwitter APIを使って自動でTweetする仕組みの構築を目指す。
Twitterのドキュメントから、まず以下のステップで認証を行いTweetを行う必要がある。OAuth1.0に従ったフローのようだ。

Request Tokenの取得
Twitterの認証
Access Tokenの取得
Tweetを実行

Request Tokenの取得

Resource URL
https://api.twitter.com/oauth/request_token
Method
Post
Authorization Header Parameters
以下のパラメータをAuthorizationヘッダーにセットする
Authorization: OAuth oauth_callback="http%3A%2F%2F127.0.0.1", oauth_nonce="...
- oauth_callback（必須）
  APIに登録済みのコールバックURL
  ユーザがTwitterの認証後にリダイレクトされる
- x_auth_access_type（オプション）
  アクセス権の設定
  readもしくはwriteを設定可能
- oauth_consumer_key（必須）
  Twitter APIのAPI Key
- oauth_signature_method（必須）
  このリクエストの署名のアルゴリズム（HMAC-SHA1など）
- oauth_signature（必須）
  署名
- oauth_timestamp（必須）
  リクエスト発行時のUNIXTIME
- oauth_nonce（必須）
  リクエスト毎に一意なランダムな値
- oauth_version（オプション）
  1.0
Request Body
なし
サンプルコード
requests_oauthlibを使えば簡単だが、勉強がてら使わずに実装してみた

get_request_token.py

import time
import base64
import os
import uuid
import hashlib
import hmac
import urllib.parse
import requests

REQUEST_URL = 'https://api.twitter.com/oauth/request_token'
REQUEST_METHOD = 'POST'
API_KEY = os.environ.get('API_KEY')
API_SECRET = os.environ.get('API_SECRET')
TOKEN_SECRET = ''
CALLBACK_URL = os.environ.get('CALLBACK_URL')
SIGNATURE_METHOD = 'HMAC-SHA1'

def _create_timestamp():
    return str(int(time.time()))

def _create_nonce():
    return str(uuid.uuid4())

def _create_base_string(params: dict):
    string = ''
    for key, val in sorted(params.items()):
        string += key + '=' + val + '&'

    return string.rstrip('&')

def _convert_dict(dict_: dict):
    string = ''
    for key, val in dict_.items():
        string += key + '="' + val + '", '

    return string.rstrip(', ')

def _create_signature(base_string: str):
    base_string = REQUEST_METHOD + '&' + urllib.parse.quote(REQUEST_URL, safe='') + '&' + urllib.parse.quote(base_string, safe='')
    key = API_SECRET + '&' + TOKEN_SECRET
    hashed_string = hmac.new(key.encode(), base_string.encode(), hashlib.sha1)
    return urllib.parse.quote(base64.standard_b64encode(hashed_string.digest()), safe='')

def _create_authorization_params():
    authorization_params = {}
    authorization_params['oauth_timestamp'] = _create_timestamp()
    authorization_params['oauth_consumer_key'] = urllib.parse.quote(API_KEY, safe='')
    authorization_params['oauth_nonce'] = _create_nonce()
    authorization_params['oauth_callback'] = urllib.parse.quote(CALLBACK_URL, safe='')
    authorization_params['oauth_signature_method'] = SIGNATURE_METHOD
    authorization_params['oauth_version'] = '1.0'
    authorization_params['oauth_signature'] = _create_signature(_create_base_string(authorization_params))

    return authorization_params

def _create_authorization_header():
    authorization_header_val = 'OAuth '
    params = _create_authorization_params()
    authorization_header_val += _convert_dict(params)

    return {'Authorization': authorization_header_val}

def execute():
    headers = _create_authorization_header()

    return requests.post(REQUEST_URL, headers=headers)

result = _execute()
print(result.text)
# -> oauth_token=xxxxxxxxxxxxxxxx&oauth_token_secret=yyyyyyyyyyyyyyyyyyyyyyyyyyyyy&oauth_callback_confirmed=true

requests_oauthlibを使った場合は以下のように簡単に書ける

get_request_token_w_authlib.py

import os
import requests
from requests_oauthlib import OAuth1Session
from urllib.parse import parse_qsl

REQUEST_URL = 'https://api.twitter.com/oauth/request_token'
REQUEST_METHOD = 'POST'
API_KEY = os.environ.get('API_KEY')
API_SECRET = os.environ.get('API_SECRET')
TOKEN_SECRET = ''
CALLBACK_URL = os.environ.get('CALLBACK_URL')
SIGNATURE_METHOD = 'HMAC-SHA1'

twitter_api = OAuth1Session(API_KEY, API_SECRET)
response = twitter_api.post(REQUEST_URL, params={'oauth_callback': CALLBACK_URL})

print(response.text)
# -> oauth_token=xxxxxxxxxxxxxxxx&oauth_token_secret=yyyyyyyyyyyyyyyyyyyyyyyyyyyyy&oauth_callback_confirmed=true

続編で上記コードをAPI Gateway + Lambdaにデプロイした記事をアップしました