今回はエクセルに予め用意したツイート文を、指定した時間にツイートしていくツールをLambdaとS3を使用して作成していきます。
ツイート文は連投もできるようにします。

前提

今回はCLIを積極的に使おうということでやっていきます。
awsCLIのインストールについてはこちらを参照してください。

Twitter APIの取得についてもこちらを参考に取得してください。

API Key
API Secret
Access Token
Access Secret
以上を取得してください。

また、python3.9を使用するのでそちらもインストールしておいてください。

手順

S3のバケットを作成
各パッケージインストール
ロールの作成
ポリシーの作成・アタッチ
関数の作成
Lambda関数の登録
環境変数の登録
トリガーの設定
S3のファイルを移動する
関数の作成
トリガーの設定

S3バケットを作成

S3のバケットを作成するのですが、ここでは2つ作成します。
後ほど使用済みファイルを別フォルダに移動する処理を書くのですが、同一バケット内で処理したときに、トリガーが無限ループする可能性を避けるためです。
以下のコマンドを実行

$ aws s3 mb s3://mybucket
make_bucket: mybucket
$ aws s3 mb s3://mybucket-done
make_bucket: mybucket-done

$ aws s3 lsで確認出来たら作成成功

次に作成したバケットの中にファイルを入れます。
月ごとに作成したファイルを取得したいので、ファイル名は2024_04.xlsxとします。

$ aws s3 cp test.xlsx s3://mybucket
upload: ./2024_04.xlsx to s3://mybucket/2024_04.xlsx

エクセルのデータは以下のようになっています。

 ['日付', 'イベント', '原文', '投稿内容1', '連投1-1', '連投1-2', '連投1-3', '投稿時間1', '画像', '原文', '投稿内容2',...以下略],
 ['1日', 'イベント','None', 'テストツイート', '連投テスト', None, None, datetime.time(7, 0), None,...以下略]

連投についての注意点
Xでは同じ内容を連投することはできないようです。
なのでテストファイルでも違う文を用意してください。

各パッケージのインストール

ディレクトリを作成し、各パッケージをインストールしてください。

mkdir sns-test
cd sns-test
pip3 install boto3 -t ./
pip install requests_oauthlib -t ./
pip install openpyxl -t ./

ロールの作成

iamフォルダを作成し、role.jsonに以下を記述する

iam/role.json


{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

以下コマンドでロールを作成し、jsonが出力されればOKです。
ロールのARNが出力されるので、控えておきます。

$ aws iam create-role --role-name <ロールの名前> --assume-role-policy-document file://iam/role.json

ポリシーの作成・アタッチ

S3のフルアクセスの権限とCloudWatch Logsの作成・書き込み権限を作成し、アタッチしていきます。
S3の権限についてはawsで用意されている既存のものを使用するので、下記のコマンドでアタッチするだけです。

$ aws iam attach-role-policy --role-name <ロールの名前> --policy-arn arn:aws:iam::aws:policy/AmazonS3FullAccess

CloudWatch Logsのjsonの作成の前に、CloudWatchのLogGroupを作成します。

$ aws logs create-log-group --log-group-name /aws/lambda/<後ほど作成する関数と同じ名前を推奨>

以下のコマンドで作成したロググループのArnを調べます。

$ aws logs describe-log-groups --log-group-name-prefix /aws/lambda/<ロググループの名前>

次に以下のiam/policy.jsonを作成します。
先ほどロググループの情報で出力された「arn」の方を「"Resource": "<作成したロググループのArn>"」に入れます。「logGroupArn」も出力されますが、間違えないように。

iam/policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "logs:CreateLogGroup",
      "Resource": "arn:aws:logs:ap-northeast-1:<自身のID>:*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "<作成したロググループのArn>"
    }
  ]
}

関数の作成

Xへ投稿するファイルとエクセルデータを取得するファイルに分けてみました。
index.pyがエントリーポイントになります。

index.py

import json
import os
from requests_oauthlib import OAuth1Session
import boto3
from get_excel_data import get_post_content_for_time, get_excel_data
from datetime import datetime, timedelta, timezone

# Twitter APIの認証情報を取得
config = json.load(open('config/env.json', 'r'))
consumer_key = config['Variables']['consumerKey']
client_secret = config['Variables']['clientSecret']
access_token = config['Variables']['accessToken']
access_token_secret = config['Variables']['accessTokenSecret']

# OAuth1Sessionを使用してTwitter APIにアクセス
oauth = OAuth1Session(consumer_key, client_secret, access_token, access_token_secret)

# タイムゾーンの定義・取得
JST = timezone(timedelta(hours=+9), 'JST')
file_name = datetime.now(JST).strftime("%Y_%m")
post_time = datetime.now(JST).strftime("%H:%M")

# S3バケット名とオブジェクトキー名を定義
BUCKET_NAME = config['Variables']['bucketName']
OBJECT_KEY_NAME = file_name + '.xlsx'

# Xへ投稿する
def handler(event, context):
    """
    Parameters:
        event: Lambdaイベント情報
        context: Lambda実行コンテキスト

    Returns:
        None
    """
    try:
        # Excelデータを取得
        excel_data = get_excel_data(BUCKET_NAME, OBJECT_KEY_NAME)
        if excel_data is not None:
            # 指定された時間の投稿内容と連投を取得する
            post_content, repeats = get_post_content_for_time(excel_data, 'X', post_time)

            if post_content is not None:
                # スペースを取り除く
                post_content = post_content.strip()
                # X投稿
                payload = {'text': post_content}
                response = oauth.post(
                    "https://api.twitter.com/2/tweets",
                    json=payload,
                )
                if response.status_code != 201:
                    raise Exception(
                        "[Error] {} {}".format(response.status_code, response.text)
                    )

                # 連投があればそれぞれをポストする
                if repeats:
                    for repeat in repeats:
                        if repeat is not None:
                            repeat = repeat.strip()
                            payload = {'text': repeat}
                            response = oauth.post(
                                "https://api.twitter.com/2/tweets",
                                json=payload,
                            )
                            if response.status_code != 201:
                                raise Exception(
                                    "[Error] {} {}".format(response.status_code, response.text)
                                )
            else :
                print("投稿時間 {} の投稿内容は見つかりませんでした。".format(post_time))
    except Exception as e:
            print("An error occurred:", e)

次にエクセルデータを取得するファイルです。

get_excel_data.py

import boto3
from openpyxl import load_workbook
from io import BytesIO
from datetime import datetime, timedelta, timezone

JST = timezone(timedelta(hours=+9), 'JST')
post_date = datetime.now(JST).strftime("%-d日")
print('日にち：', post_date)

# Excelファイルから特定の投稿内容を取得
def get_post_content_for_time(excel_data,sheet_name, post_time):
    """
    Parameters:
        excel_data (openpyxl.workbook.Workbook): Excelデータ
        sheet_name (str): シート名
        post_time (str): 投稿時間（HH:MM）

    Returns:
        tuple: 投稿内容と連投リストのタプル
    """
    sheet = excel_data[sheet_name]
    for row in sheet.iter_rows(min_row=2, values_only=True):
        if row[0] == post_date:  # 日付が一致する場合
            for i in range(7, len(row), 7):  # 投稿時間の列をループでチェック
                if row[i].strftime("%H:%M") == post_time:  # 投稿時間が指定された時間の場合
                    post_content = row[i - 4]  # 対応する投稿内容を取得
                    print("投稿内容：", post_content)
                    repeats = [r for r in row[i-3:i] if r]  # 連投のリストを取得（空でない値）
                    print("連投内容：", repeats)
                    return post_content.strip(), repeats  # 投稿内容と連投リストを返す（スペースを取り除く）
    return None, None  # 該当する投稿内容が見つからない場合はNoneを返す

# Excelファイルを取得する関数
def get_excel_data(BUCKET_NAME, OBJECT_KEY_NAME):
    """
    Parameters:
        BUCKET_NAME (str): S3バケット名
        OBJECT_KEY_NAME (str): オブジェクトキー名

    Returns:
        openpyxl.workbook.Workbook: Excelデータ
    """
    try:
        s3_client = boto3.client('s3')
        res = s3_client.get_object(Bucket=BUCKET_NAME, Key=OBJECT_KEY_NAME)
        excel_data = load_workbook(BytesIO(res['Body'].read()), data_only=True)
        return excel_data
    except Exception as e:
        print("An error occurred while loading the Excel file:", e)
        return None

Lambda関数の登録

完成したコードをzipファイルにまとめます。

$ zip -r sns-test.zip ./

以下のコマンドを実行して関数を登録します。
先ほども記述したように、関数の名前はロググループと同一が望ましいです。

aws lambda create-function --function-name <作成する関数の名前> --role <作成したロールのARN> --region ap-northeast-1 --zip-file fileb://sns-test.zip --runtime python3.9 --handler index.handler --timeout 30

オプションについては公式で詳しく説明してます。
https://docs.aws.amazon.com/cli/latest/reference/lambda/create-function.html

環境変数の登録

次に環境変数を登録します。

config/env.json

{
  "Variables": {
    "consumerKey": "API key",
    "consumerSecret": "API key seacret",
    "accessToken": "Access Token",
    "accessTokenSecret": "Access Token Secret",
    "bucketName": "mybucket"
  }
}

下記コマンドで登録をします。

aws lambda update-function-configuration --function-name <作成した関数の名前> --environment file://config/env.json

トリガーの設定

ルールの作成

月～金の7:00～21:00まで1時間ごとに設定します。
UTC時間で設定が必要なので、JSTだと+9時間の計算で計算します。

$ aws events put-rule --name "sns-post-1" --schedule-expression "croncron(0/60 0-12 ? * MON-FRI *)" --state ENABLED

$ aws events put-rule --name "sns-post-2" --schedule-expression "cron(0/60 22-23 ? * SUN-THU *)" --state ENABLED

↑単純に7:00～21:00だと12:00～22:00になるので(12-22 SUN-FRI)としてもいいのですが、そうするとローカルタイムに直したときに土日の余計な時間にも回ってしまいます。関数の実行もタダではないので、、、

ターゲットの追加

$ aws events put-targets --rule "sns-post-1" --targets Arn=関数のArn,Id=1

$ aws events put-targets --rule "sns-post-2" --targets Arn=関数のArn,Id=2

トリガーの追加

$ aws lambda add-permission --function-name 関数名 --statement-id 1 --action "lambda:InvokeFunction" --principal events.amazonaws.com --source-arn "ルールのArn"

$ aws lambda add-permission --function-name 関数名 --statement-id 2 --action "lambda:InvokeFunction" --principal events.amazonaws.com --source-arn "ルールのArn"

以上でツイートのlambda関数は完成です。

S3間でファイルを移動する

ここからは使用したファイルを使用済バケットへ移動する関数を別で作成します。以下に関しては手順を省きます。

別バケットにした理由
バケットにPutしたことをトリガーにした時に、同一バケット内で移動させると、移動したときにもイベント発火してしまうことで無限ループになってしまうそうです。
今回はループに陥るトリガー設定ではないのですが、間違いがあると悲惨なので別バケットに移動するようにしました。

ロールの作成
ロググループの作成
ポリシーの作成・アタッチ
環境変数の作成

インストールするパッケージはboto3のみです。

$ mkdir s3-sns-file-move
$ pip3 install boto3 -t ./

index.py

import boto3
import json
from datetime import datetime, timezone, timedelta
from dateutil.relativedelta import relativedelta

JST = timezone(timedelta(hours=+9), 'JST')
previous_month = datetime.now(JST) - relativedelta(months=1)
file_name = previous_month.strftime("%Y_%m")
year = previous_month.strftime("%Y")

config = json.load(open('config/env.json', 'r'))
SOURCE_BUCKET_NAME = config['Variables']['sourceBucketName']
SOURCE_OBJECT_KEY_NAME = config['Variables']['sourceObjectKeyName']
DESTINATION_BUCKET_NAME = config['Variables']['DestinationBucketName']

s3_client = boto3.client('s3')

def handler(event, context):
    try:
        destination_object_key_name = f"{year}/{SOURCE_OBJECT_KEY_NAME}"
        # 移動先のキーが存在しない場合は作成する
        try:
            s3_client.head_object(Bucket=DESTINATION_BUCKET_NAME, Key=destination_object_key_name)
        except s3_client.exceptions.ClientError as e:
            if e.response['Error']['Code'] == '404':
                # 新しいオブジェクトを追加する
                s3_client.put_object(Bucket=DESTINATION_BUCKET_NAME, Key=destination_object_key_name, Body='')
                print(f"Created new destination key: {destination_object_key_name}")
            else:
                raise e

        # S3からファイルを移動する
        s3_client.copy_object(
            Bucket=DESTINATION_BUCKET_NAME,
            CopySource={'Bucket': SOURCE_BUCKET_NAME, 'Key': SOURCE_OBJECT_KEY_NAME},
            Key=destination_object_key_name
        )

        # 元のファイルを削除する
        s3_client.delete_object(
            Bucket=SOURCE_BUCKET_NAME,
            Key=SOURCE_OBJECT_KEY_NAME
        )
        print("S3ファイルの移動と削除が完了しました。")
    except Exception as e:
        print("An error occurred:", e)

関数の作成

$ aws lambda create-function --function-name 関数名 --role ロールのArn --region ap-northeast-1 --zip-file fileb://s3-sns-file-move.zip
--runtime python3.9 --handler index.handler --timeout 10

トリガーの設定

ルールの作成

今回使用するエクセルのファイルは月ごとなので、毎月１日に移動するように設定します。

$ aws events put-rule --name "s3-sns-move-file" --schedule-expression "cron(0 0 1 * ? *)" --state ENABLED
{
    "RuleArn": "arn:aws:events:ap-northeast-1:047445725132:rule/s3-sns-move-file"
}

ターゲットの追加

$ aws events put-targets --rule "s3-sns-move-file" --targets Arn=arn:aws:lambda:ap-northeast-1:047445725132:function:s3-test,Id=1

トリガーの追加

$ aws lambda add-permission --function-name s3-test --statement-id 1 --action "lambda:InvokeFunction" --principal events.amazonaws.com --source-arn "arn:aws:events:ap-northeast-1:047445725
132:rule/s3-sns-move-file"
{
    "Statement": "{\"Sid\":\"1\",\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"events.amazonaws.com\"},\"Action\":\"lambda:InvokeFunction\",\"Resource\":\"arn:aws:lambda:ap-northeast-1:047445725132:function:s3-test\",\"Condition\":{\"ArnLike\":{\"AWS:SourceArn\":\"arn:aws:events:ap-northeast-1:047445725132:rule/s3-sns-move-file\"}}}"
}

以上です。

エラー

初っ端にawsコマンドを叩いたらエラーが発生。

$ aws s3 ls
An error occurred (RequestTimeTooSkewed) when calling the ListBuckets operation: The difference between the request time and the current time is too large.

実行環境のシステム時間と現在時間が大きくズレていると発生するようです。
現在時刻を確認。

$ timedatectl status
               Local time: Wed 2024-02-07 01:44:14 JST
           Universal time: Tue 2024-02-06 16:44:14 UTC
                 RTC time: Tue 2024-02-06 16:44:15
                Time zone: Asia/Tokyo (JST, +0900)
System clock synchronized: no
              NTP service: inactive
          RTC in local TZ: no

時間を変更。

$ timedatectl set-time "2024-02-07 13:50:00"
$ timedatectl status
               Local time: Wed 2024-02-07 13:50:02 JST
           Universal time: Wed 2024-02-07 04:50:02 UTC

もしタイムゾーンが違う方がいたらタイムゾーンも直しましょう。

$ timedatectl set-timezone Asia/Tokyo