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

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

1. Request Tokenの取得
2. Twitterの認証
3. Access Tokenの取得
4. 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にデプロイした記事をアップしました