AWS Lambda(Python)からTwitterに投稿する

2020/05/11

「GoogleフォームからAPI Gatewayで作成したREST APIにPOSTリクエストする」の続きです。

この部分を作ります。

「PythonでTwitter API を利用していろいろ遊んでみる」を参考にさせていただきました。

Twitterアプリケーション(認証情報)の取得

対象アカウントでTwitterにログインして、こちらhttps://developer.twitter.com/en/appsにアクセスしました。

[Create an app]を押下しました。

なんで使うんだ？と聞かれたようなので、「趣味でボット作成」としました。

電話番号の紐付けが必要なようですので、紐付けして画面更新して、下の方で国を選択して、[Next]ボタンを押下しました。

Twitter APIを使う理由を200文字以上で英語で書いて、とありましたので、Google翻訳に頼って書きました。

The specifics(詳細)セクションは1つだけYesでした。
「ツイート機能を使いますか？使うならどう使いますか？」と来たので、上記と同じことを書きました。

申請が終わったようです。
メールが届いて、[Confirm]しました。

もう一度、https://developer.twitter.com/en/appsにアクセスして、[Create an app]ボタンを押下しました。

アプリ名、内容、URLが必須でした。

あと、使い方も必須でした。

[Keys and tokens]タブに、API keyとAPI secret keyがありました。
[Generate]ボタンで、Access token とAccess token secretも発行しました。

AWS Secrets Mangerの準備

Twitter APIの4つの認証情報は、AWS Secrets Managerで管理することにしました。

[新しいシークレットを保存する]ボタンを押下しました。

シークレットの種類は、「その他のシークレット（APIキーなど）」を選択しました。

Twitter API の4つのキーをキーバリューで入力します。
暗号化キーはデフォルトのままにしておきました。

名前とタグを設定しました。

ローテーションは無効にしました。
そのうち、トークンの再発行をするLambdaとかを動かすかもですが。

できました。
サンプルコード付きです。
親切です。

AWS Lambda Layersの準備

requests とrequests-oauthlib のLayersがあると便利そうなので、まとめて作りました。

開発環境のmacで作業しました。

$ mkdir python
$ pip install requests requests-oauthlib -t python
$ zip -r python.zip python/

$ mkdir python

$ pip install requests requests-oauthlib -t python

$ zip -r python.zip python/

pythonディレクトリごと、zipにして、Layersにアップロードしました。

IAMロール

Lambda用のIAMロールとして、AWSLambdaBasicExecutionRoleと、以下のインラインポリシーを設定しました。

SecretsManagerから認証情報を取得するためのポリシー

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "secretsmanager:GetSecretValue",
            "Resource": "arn:aws:secretsmanager:us-east-1:123456789012:secret:LiveDelivery/TwitterAPIKey-*****"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Effect": "Allow",

"Action": "secretsmanager:GetSecretValue",

"Resource": "arn:aws:secretsmanager:us-east-1:123456789012:secret:LiveDelivery/TwitterAPIKey-*****"

}

]

}

SQSキューをトリガーとして設定するためのポリシー

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "sqs:ReceiveMessage",
                "sqs:DeleteMessage",
                "sqs:GetQueueAttributes"
            ],
            "Resource": "arn:aws:sqs:us-east-1:123456789012:LiveTweet"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Effect": "Allow",

"Action": [

"sqs:ReceiveMessage",

"sqs:DeleteMessage",

"sqs:GetQueueAttributes"

"Resource": "arn:aws:sqs:us-east-1:123456789012:LiveTweet"

}

]

}

Amazon SNS

SNSのサブスクリプションのSQSは、raw メッセージ配信を有効にしました。

AWS Lambda

事前に準備した、Layers、 IAMロールを設定して、トリガーはSQSにしました。

環境変数は2つ作成しました。
* SECRET_NAME: Secrets Mangerのシークレット名
* LOG_LEVEL: ログレベル

import json
import os
import boto3
import base64
import logging
import traceback
from botocore.exceptions import ClientError
from requests_oauthlib import OAuth1Session

secret_name = os.environ.get('SECRET_NAME')
level = os.environ.get('LOG_LEVEL', 'ERROR')

def logger_level():
    if level == 'CRITICAL':
        return 50
    elif level == 'ERROR':
        return 40
    elif level == 'WARNING':
        return 30
    elif level == 'INFO':
        return 20
    elif level == 'DEBUG':
        return 10
    else:
        return 0

def get_secret():

    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager'
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )
    except ClientError as e:
        raise e
    else:
        if 'SecretString' in get_secret_value_response:
            secret = get_secret_value_response['SecretString']
        else:
            secret = base64.b64decode(get_secret_value_response['SecretBinary'])

    return secret

def lambda_handler(event, context):
    logger = logging.getLogger()
    logger.setLevel(logger_level())

    logger.debug(event)

    try:
        secret = json.loads(get_secret())
        logger.debug(secret)

        twitter = OAuth1Session(
            secret['api_key'],
            secret['api_secret_key'],
            secret['access_token'],
            secret['access_token_secret']
        )
        url = "https://api.twitter.com/1.1/statuses/update.json"

        for record in event['Records']:
            message = json.loads(record['body'])
            tweet = '【配信ライブカレンダー新着情報】\n'\
                    'https://www.yamamanx.com/delivery_live_calendar/\n'\
                    '{title}\n{startTime}~\n'\
                    '{description}'.format(
                        title=message['title'],
                        startTime=message['startTime'],
                        description=message['description']
                    )
            params = {
                'status': tweet[:140]
            }

            res = twitter.post(url, params = params)
            logger.debug(res)

            return {
                'statusCode': res.status_code
            }

    except Exception as e:
        logger.error(traceback.format_exc())
        raise e

import json

import os

import boto3

import base64

import logging

import traceback

from botocore.exceptions import ClientError

from requests_oauthlib import OAuth1Session

secret_name = os.environ.get('SECRET_NAME')

level = os.environ.get('LOG_LEVEL', 'ERROR')

def logger_level():

if level == 'CRITICAL':

return 50

elif level == 'ERROR':

return 40

elif level == 'WARNING':

return 30

elif level == 'INFO':

return 20

elif level == 'DEBUG':

return 10

else:

return 0

def get_secret():

session = boto3.session.Session()

client = session.client(

service_name='secretsmanager'

)

try:

get_secret_value_response = client.get_secret_value(

SecretId=secret_name

)

except ClientError as e:

raise e

else:

if 'SecretString' in get_secret_value_response:

secret = get_secret_value_response['SecretString']

else:

secret = base64.b64decode(get_secret_value_response['SecretBinary'])

return secret

def lambda_handler(event, context):

logger = logging.getLogger()

logger.setLevel(logger_level())

logger.debug(event)

try:

secret = json.loads(get_secret())

logger.debug(secret)

twitter = OAuth1Session(

secret['api_key'],

secret['api_secret_key'],

secret['access_token'],

secret['access_token_secret']

)

url = "https://api.twitter.com/1.1/statuses/update.json"

for record in event['Records']:

message = json.loads(record['body'])

tweet = '【配信ライブカレンダー新着情報】\n'\

'https://www.yamamanx.com/delivery_live_calendar/\n'\

'{title}\n{startTime}~\n'\

'{description}'.format(

title=message['title'],

startTime=message['startTime'],

description=message['description']

)

params = {

'status': tweet[:140]

}

res = twitter.post(url, params = params)

logger.debug(res)

return {

'statusCode': res.status_code

}

except Exception as e:

logger.error(traceback.format_exc())

raise e

【配信ライブカレンダー新着情報】https://t.co/Fbk1Bd0TRD
眉村ちあきメジャーデビュー1周年記念ライブ
2020/05/07 18:30:00~
高い歌唱力に加えギターを弾き語りするスタイル、さらにユーモアの

— live.calendar.haishin (@LHaishin) May 10, 2020

できました！

最後までお読みいただきましてありがとうございました！

【PR】「AWS認定資格試験テキスト＆問題集　AWS認定ソリューションアーキテクト - プロフェッショナル改訂第2版」という本を書きました。

【PR】「AWS認定資格試験テキスト AWS認定クラウドプラクティショナー改訂第2版」という本を書きました。

【PR】「ポケットスタディ AWS認定デベロッパーアソシエイト」という本を書きました。

【PR】「要点整理から攻略するAWS認定ソリューションアーキテクト-アソシエイト」という本を書きました。

【PR】「AWSではじめるLinux入門ガイド」という本を書きました。