LoginSignup
1
1

More than 3 years have passed since last update.

@jnakanaka1

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

Last updated at Posted at 2019-06-08

今までの内容は以下の通り

今回は、取得したAccess Tokenを使って定期的にTweetする仕組みを作る。まず、Access TokenをDynamoDBに保存する。そして、CloudWatch Eventを使って定期的にLambdaを実行して、DynamoDBからAccess Tokenを取得してTweetする。

DynamoDB Tableの作成

DynamoDBコンソールにアクセスし、テーブルを作成する。テーブル名は「token-table」、プライマリーキーは「id」として、「作成」をクリックする。

dynamodb-1.PNG

DynamoDBアクセス用Lambda Layerの作成

DynamoDBにtokenを保存したり、保存したtokenを取得するために、DynamoDBにアクセスする関数を作成し、Layerに登録する。Layerに登録することで複数のLambda関数からLayerを使ってDynamoDBにアクセスすることができる。以下はDynamoDBへアクセスするためのコード。Layerの使い方については「Twitter APIを使って自動でTweetする（５）」を参照してほしい。

dynamodb.py
import boto3

class TokenTable:
    parition_name = 'id'
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('token-table')

    @classmethod
    def register(cls, data):
        cls.table.put_item(Item=data)

    @classmethod
    def get_item(cls, id):
        response = cls.table.get_item(
            Key={
                'id': id
            }
        )

        return response.get('Item')

また、LambdaからDynamoDBにアクセスするためにLambdaに権限を追加する必要がある。
DynamoDBにアクセスする必要のあるget_access_token関数を開き、「IAMコンソールでget_access_token-xxxxロールを表示します。」をクリックする。
dynamodb-2.PNG
「インラインポリシーの追加」をクリックする。
IAM-1.PNG
「JSON」タブをクリックする。
IAM-2.PNG
インラインポリシーをJSON形式で入力する。
IAM-3.PNG

ポリシーは以下のように入力し、「ポリシーの確認」をクリックする。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "dynamodb:*",
            "Resource": "arn:aws:dynamodb:*:*:table/token-table"
        }
    ]
}

ポリシー名を入力し、「ポリシーの作成」をクリックする。
IAM-4.PNG
次に、get_access_token関数を以下のように修正し、取得したAccess TokenをDynamoDBのtoken-tableに保存するように変更する。

get_access_token.py
import os
import urllib.parse
import requests
import utilities
from dynamodb import TokenTable

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

class GetAccessToken:

    def __init__(self, event, context):
        self.oauth_token = event.get('queryStringParameters').get('oauth_token')
        self.oauth_verifier = event.get('queryStringParameters').get('oauth_verifier')

    def _create_authorization_params(self):
        authorization_params = {}
        authorization_params['oauth_timestamp'] = utilities.create_timestamp()
        authorization_params['oauth_consumer_key'] = urllib.parse.quote(API_KEY, safe='')
        authorization_params['oauth_nonce'] = utilities.create_nonce()
        authorization_params['oauth_signature_method'] = SIGNATURE_METHOD
        authorization_params['oauth_version'] = '1.0'
        authorization_params['oauth_token'] = self.oauth_token
        authorization_params['oauth_verifier'] = self.oauth_verifier
        authorization_params['oauth_signature'] = utilities.create_signature(utilities.create_base_string(authorization_params), REQUEST_METHOD, REQUEST_URL, API_SECRET, TOKEN_SECRET)

        return authorization_params

    def _create_authorization_header(self):
        authorization_header_val = 'OAuth '
        params = self._create_authorization_params()
        authorization_header_val += utilities.convert_dict(params)

        return {'Authorization': authorization_header_val}

    def _parse_response(self, response):
        response_dict = {
            (response.text.split('&')[0]).split('=')[0]: (response.text.split('&')[0]).split('=')[1],
            (response.text.split('&')[1]).split('=')[0]: (response.text.split('&')[1]).split('=')[1],
            (response.text.split('&')[2]).split('=')[0]: (response.text.split('&')[2]).split('=')[1],
            (response.text.split('&')[3]).split('=')[0]: (response.text.split('&')[3]).split('=')[1]
        }
        data = {
            'id': response_dict['user_id'],
            'token': response_dict['oauth_token'],
            'secret': response_dict['oauth_token_secret'],
            'name': response_dict['screen_name']
        }
        return data

    def _register_id(self, data):
        token_table = TokenTable()
        token_table.register(data)

    def _execute(self):
        headers = self._create_authorization_header()
        response = requests.post(REQUEST_URL, headers=headers)
        data = self._parse_response(response)
        self._register_id(data)

def lambda_handler(event, context):

    get_access_token = GetAccessToken(event, context)
    get_access_token._execute()

    return {
        'statusCode': 200,
        'headers': {'Access-Control-Allow-Origin': '*'},
        'body': 'Registration Success!'
    }

これで、Access TokenをDynamoDBに保存されるようになった。あとは、このAccess Tokenを使ってTweetする関数を作成し、CloudWatch Eventを使って定期的にそのLambdaを実行すれば自動でTweetすることができる。

Tweetする関数のコードは以下のように作成した。環境変数、ロール、レイヤーはget_access_tokenの関数と同様に設定している。
また、token-tableにアクセスする際のパーティションキーであるidの値を今回は決め打ちで設定している。この値は自身のtoken-tableを確認し、保存されているidに書き換えて使う必要がある。

auto_tweet.py
import os
import urllib.parse
import requests
import utilities
from dynamodb import TokenTable

REQUEST_METHOD = 'POST'
API_KEY = os.environ.get('API_KEY')
API_SECRET = os.environ.get('API_SECRET')
SIGNATURE_METHOD = 'HMAC-SHA1'

TWEET_URL = 'https://api.twitter.com/1.1/statuses/update.json'

class AutoTweet:

    def __init__(self, event, context):
        pass

    def get_token(self, id):
        token_table = TokenTable()
        res = TokenTable.get_item(id)
        return {'token': res.get('token'), 'secret': res.get('secret')}

    def _create_authorization_params_for_tweet(self, token, token_secret, message):
        signature_params = {}
        signature_params['status'] = urllib.parse.quote(message, safe='')

        authorization_params = {}
        authorization_params['status'] = urllib.parse.quote(message, safe='')
        authorization_params['oauth_timestamp'] = utilities.create_timestamp()
        authorization_params['oauth_consumer_key'] = urllib.parse.quote(API_KEY, safe='')
        authorization_params['oauth_nonce'] = utilities.create_nonce()
        authorization_params['oauth_version'] = '1.0'
        authorization_params['oauth_signature_method'] = SIGNATURE_METHOD
        authorization_params['oauth_token'] = token

        signature_params.update(authorization_params)

        authorization_params['oauth_signature'] = utilities.create_signature(utilities.create_base_string(signature_params), REQUEST_METHOD, TWEET_URL, API_SECRET, token_secret)

        return authorization_params

    def _create_authorization_header_for_tweet(self, token, token_secret, message):
        authorization_header_val = 'OAuth '
        params = self._create_authorization_params_for_tweet(token, token_secret, message)
        authorization_header_val += utilities.convert_dict(params)

        return {'Authorization': authorization_header_val}

    def _tweet(self, token, token_secret, message):
        ENDPOINT = TWEET_URL + '?status=' + urllib.parse.quote(message, safe='')
        headers = self._create_authorization_header_for_tweet(token, token_secret, message)
        return requests.post(ENDPOINT, headers=headers)

    def _execute(self):
        id = '1132669739304513536'
        token_info = self.get_token(id)

        return self._tweet(token_info['token'], token_info['secret'], 'Hello,  this is auto tweet test:' + utilities.create_timestamp())

def lambda_handler(event, context):

    auto_tweet = AutoTweet(event, context)
    try:
        auto_tweet._execute()

    except Exception as e:

        raise e

    return {
        'statusCode': 200,
        'headers': {'Access-Control-Allow-Origin': '*'},
        'body': 'success!'
    }

最後に、CloudWatch Eventの設定をする。
まず、CloudWatchコンソールにアクセスし、イベントのルールからルール作成をクリックする。
CloudWatch-1.png
イベントソースでスケジュールを選択し、適当な値を設定してターゲットで作成したLambda関数を選択して、「設定の詳細」をクリックする。
CloudWatch-2.png
名前を入力し、状態を「有効」にして「ルールの作成」をクリックする。
CloudWatch-3.png

これで、「Twitter APIを使って自動でTweetする(3)」で作成したVueアプリからログインをすると、そのログインしたTwitterのアカウント（実際は、上記で指定したidに紐づくアカウント）で定期的にTweetされるようになる。

次回はフロントエンドを追加して、汎用的に使えるようにする予定。

前回までの投稿は以下
Twitter APIを使って自動でTweetする(1)
Twitter APIを使って自動でTweetする(2)
Twitter APIを使って自動でTweetする(3)
Twitter APIを使って自動でTweetする(4)
Twitter APIを使って自動でTweetする(5)

1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
1