モチベーション

写真を撮る目的は様々です。私の場合は幸せな時間などを思い出し、今を生きる力にすることが目的です。思い出は私に愛と勇気、そして圧倒的なパワーを与えます。しかし、ある日私は気づきました。「あれ、写真たくさん撮ってるけど見る頻度少なくない、、、？？」

データを溜めただけで満足してはいけません。どのように活用するかが重要なのです。そこで私は毎日チェックするTwitterで思い出を振り返れるようにしたのです。

アーキテクチャ概要

Cloud Schedulerにより１時間ごとにCloud Functionsが実行され、Tweetをします。ランタイムはPythonです。Cloud Functionsは実行されると、Google Photoから写真を取得し、Twitterに画像付きでツイートをします。Google PhotoやTwitterで利用するクレデンシャルはSecret Managerに保存します。一部の初期構築以外は全てTerraformで管理できます。

これまでからの変更点

実はこの仕組みは2018年から実装していました。

最初はラズパイ上で動かし、次にAWS Lambdaで動かし、そして今回はCloud Functions上で動かしています。普段はインフラ系の業務が多くコーディングをしないので、「プログラミング楽しい〜」と言いながら作っています笑

今回こだわった部分を一部ご紹介します。

Terraformで構築

インフラは基本IaCすべきだと思っているので、Terraformで構築をしています。（一部Terraformで構築できない部分は手動で構築しています。）

Cloud FunctionsはコードをCloud Storageに一度アップロードしたのちにデプロイをする必要があるので工夫が必要になります。Deploy Cloud Functions on GCP with Terraform を参考にしました。

# Cloud FunctionsのコードをZIPにします。
data "archive_file" "memory_tweet_source" {
  type        = "zip"
  source_dir  = "../src"
  output_path = "/tmp/function.zip"
  excludes = [
    "venv",
    "__pycache__",
    ".gitignore",
    "client_secret.json",
  ]
}

# 作ったZIPをCloud Storageにuploadします。
resource "google_storage_bucket_object" "memory_tweet_zip" {
  source       = data.archive_file.memory_tweet_source.output_path
  content_type = "application/zip"

  # ファイル内容からMD5を作成し、ファイル名に含めます。
  # これでソースコードの変更時にCloud Funcationsが更新されるようになります。
  name   = "src-${data.archive_file.memory_tweet_source.output_md5}.zip"
  bucket = google_storage_bucket.function_bucket.name
}

# クラウドスケジューラーにより実行されるCloud Funcationを作成。
resource "google_cloudfunctions_function" "memory_tweet" {
  name    = "memory_tweet"
  runtime = "python39"

  available_memory_mb = 256

  # Cloud Functionのコードが入ったZIPを指定します。
  source_archive_bucket = google_storage_bucket.function_bucket.name
  source_archive_object = google_storage_bucket_object.memory_tweet_zip.name

  entry_point = "memory_tweet"

  event_trigger {
    event_type = "google.pubsub.topic.publish"
    resource   = google_pubsub_topic.memory_tweet_trigger.id
  }

  service_account_email = google_service_account.memory_tweet_function.email

  environment_variables = {
    PROJECT_ID                          = var.project_id
    GOOGLE_OAUTH_CREDENITIALS_SECRET_ID = google_secret_manager_secret.google_oauth_credentials.secret_id
    TWITTER_CREDENITIALS_SECRET_ID      = google_secret_manager_secret.twitter_credentials.secret_id
  }
}

Tweepyを利用する。

TweepyはTwitter APIを利用するためのライブラリです。当時私はこんな便利なものがあるのを知らず、頑張ってrequests_oauthlibを使ってAPIを利用していました笑

だいぶコードがスッキリしたので良かったです。また、以前まではツイート時にUTCの時刻をメッセージに付けていましたが、今回はJSTに変換するようにしました。

Before

URL_MEDIA = "https://upload.twitter.com/1.1/media/upload.json"
URL_TEXT = "https://api.twitter.com/1.1/statuses/update.json"

def uploadMedia(mediaItem):
    # OAuth認証 セッションを開始
    env = os.environ
    twitter = OAuth1Session(env['TWITTER_CK'],env['TWITTER_CS'],env['TWITTER_AT'],env['TWITTER_AS'])

    # 画像投稿
    files = {"media": mediaItem['imageBinary']}
    req_media = twitter.post(URL_MEDIA, files=files)

    # レスポンスを確認
    if req_media.status_code != 200:
        logging.error("画像アップデート失敗: %s", req_media.text)
        sys.exit(1)

    # Media ID を取得
    media_id = json.loads(req_media.text)['media_id']
    logging.debug("Media ID: %d" % media_id)

    # Media ID を付加してテキストを投稿
    params = {'status': mediaItem['mediaMetadata']['creationTime'], "media_ids": [media_id]}
    req_media = twitter.post(URL_TEXT, params=params)

    # レスポンスを確認
    if req_media.status_code != 200:
        print("テキストアップデート失敗: %s", req_media.text)
        sys.exit(1)
    
    return

After

def uploadMedia(mediaItem, credentials):
    """
    写真付きで日付をツイートをする。
    """

    # APIの認証
    auth = tweepy.OAuthHandler(
        credentials['TWITTER_CK'], credentials['TWITTER_CS'])
    auth.set_access_token(credentials['TWITTER_AT'], credentials['TWITTER_AS'])

    api = tweepy.API(auth)

    # JSTに変換する。
    t_delta = datetime.timedelta(hours=9)  # 9時間
    dt_utc = datetime.datetime.strptime(
        mediaItem['mediaMetadata']['creationTime'], '%Y-%m-%dT%H:%M:%SZ')
    dt_jst = dt_utc + t_delta
    message = dt_jst.strftime('%Y-%m-%d %H:%M:%S %Z')

    with tempfile.NamedTemporaryFile() as f:
        f.write(mediaItem['imageBinary'])
        api.update_status_with_media(filename=f.name, status=message)

認証情報はSecret Managerに保存する

AWS Lambdaで作成した際は、セキュリティ上好ましくないですが環境変数に設定していました。今回はSecret Managerに保存し、利用しています。

今回の使い方だとリフレッシュトークンが無効になるケースは基本的にはないので同じものを使い続ければ良いのかもしれませんが、更新後の認証情報をSecret Managerに再度保存しています。

まとめ

今日の日付と同じ過去の画像をランダムにGoogle Photoから取得し、ツイートするBotを作りました。思い出から圧倒的なパワーを毎日もらえるようになります。興味があれば、試してみてください。

Paper2 Blog

ともに、かける

Google Photoの思い出をTweetするBotを作った