2021-03-12

顔認識 (OpenCV DNN) とツイート (Tweepy) をサーバーレス環境 (GCP) で定期実行

Twitterのホームタイムラインを取得して, 顔認識とツイートを定期的に実行する仕組みをサーバーレス環境 (GCP) に実装しました.
(意訳：推しの画像をリツイートしてみんなに見てもらいたかった@pedigree_C)

推しの方々からいいねをもらったりもらわなかったり, 評判がなかなか良いです(?)

全体のソースコードと詳細は https://github.com/camberbridge/dft-gcp-serverless にありますのでよろしければご覧ください.

アーキテクチャ

顔認識&ツイート定期実行アーキテクチャ

何と言っても特筆すべきは Cloud Functions の柔軟さです.

関数が何らかの原因でクラッシュしても, 「失敗時に再試行」の設定をしておけば最大7日間 (2021年現在), 関数が正常に完了するまで繰り返し呼び出されます (タイムアウト時間は設定可能).

また, リクエストの多さによってインスタンス数が自動でスケールします. もちろんこの時の上限数は設定可能です.
今回は, 毎分の小さな定期実行なのでスケーリングしなくても間に合うだろうと思っていましたが, 画像付きツイートが多いと (顔認識があるので) 関数の処理に時間がかかって次のリクエスト処理に間に合わないことが多々ありました.

そんなときに Cloud Functions は自動でスケールアウトして, 逆に画像付きツイートが少ないとスケールインしてくれて, 非常に便利でした.

サーバーレス環境 (GCP)

DBにCloud Datastore

Twitterのホームタイムラインを定期的に取得する際, 前回取得したTweet IDよりも新しいものだけを取得するようにしたかったので, 最新のTweet IDを保持しておく必要がありました.
そこで, (DBを用意するまでもない構成ですが) 環境をステートレスにしたかったこともあり, ドキュメントDBの Cloud Datastore を用いました$^1$.

Tweet IDを書き込むときはデータを増やす理由がないので同じエンティティを更新し続けるようにしました.
ベストプラクティス的には, 「1つのエンティティをあまり頻繁に (1秒に1回以上) 更新しないこと」とありますが, ここでは毎分の更新なので良しとします$^2$.

from google.cloud import datastore

datastore_client = datastore.Client()
# エンティティの種類
kind = "Twitter_home"
# エンティティの名前/ID
name = "latest_tweet_id"
entity_key = datastore_client.key(kind, name)

def put_latest_tweet_id(tweet_id):
    entity = datastore.Entity(key=entity_key)
    entity["id"] = tweet_id

    # エンティティ保存
    datastore_client.put(entity)

def get_latest_tweet_id():
    # エンティティ取得
    entity = datastore_client.get(key=entity_key)

    return entity["id"]

※ローカル環境から Datastore にアクセスする (上のコードを実行する) 場合はサービスアカウントを発行する必要があります$^3$

プログラムの実行環境にCloud Functions

イベント単位の実行に適したサーバーレスコンピューティングリソースの Cloud Functions を用いました.
基本的にはGCPコンソールで作成と設定が可能ですが, 顔認識に必要なOpenCVのDNNモデル等のファイルをコンソールからアップロードできません.

そこで, ソースコードやアセットをzip化したファイルを gsutil で Cloud Storage にアップロードしておき, その後の gcloud による Functions デプロイ時に, オブジェクトURLを指定してあげます$^4$.

$ zip -r [FILE_NAME].zip *
$ gsutil mb -p [GCP_PROJECT_ID] -l US-CENTRAL1 gs://[BUCKET_NAME]
$ gsutil cp [FILE_NAME].zip gs://[BUCKET_NAME]
$ gcloud functions deploy [FUNCTION_NAME] \
  --source=gs://[BUCKET_NAME]/[FILE_NAME].zip \
  --stage-bucket=[BUCKET_NAME] \
  --trigger-topic=[Pub/Sub_TOPIC_NAME] \
  --memory=256MB \
  --runtime=python37 \
  --region=us-central1 \
  --project=[GCP_PROJECT_ID] \
  --entry-point=main

定期実行のトリガーにCloud Scheduler x Pub/Sub

cronと同じ要領で使える Cloud Scheduler から毎分 Pub/Sub トピックをパブリッシュして Cloud Functions をトリガーさせました.
トピックは gcloud コマンドでデプロイする時に作成 (Pub/Sub_TOPIC_NAME の部分) しており, 受けの Functions にも設定しているので, あとは出す側の Cloud Scheduler の設定をします.
やり方は, GCPコンソールでリージョンやcron, トピックを指定するだけです$^5$.

その他

デプロイ時やプログラム実行時の成功/失敗など, ログ監視に Cloud Logging が大変役に立ちます.
また, エラー確認には, Cloud Error Reporting がすごく便利です. Twitter APIにレート制限があることをすっかり忘れていて, 実際に Error Reporting からメールが届いたおかげで気づけました (後述の wait_on_rate_limit を設定したきっかけ).

Twitter API (Tweepy)

レート制限の対処

無料アカウントの場合, ホームタイムライン取得リクエストの制限回数は15分に15回 (2021年現在) です$^6$.
結構シビアですが, Tweepyにはレート制限が補充されるのを自動的に待つ (エラーハンドリングせずに放置できる！) 便利なパラメータ wait_on_rate_limit があるのでそれを使います$^7$.

import tweepy

auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
api = tweepy.API(auth, wait_on_rate_limit=True)

ホームタイムライン取得

今回の仕組みは定期実行させるので, 毎度同じツイートを取得する可能性があります.
そこで, これまた便利なパラメータ since_id を使い, あるIDよりも新しいツイートを取得するようにします.$^7$

1	timeline = api.home_timeline(since_id=TWEET_ID, count=N)

ツイート前処理とMedia URLの取得

推しが投稿するツイートだけを取得したいので, 前処理として”RT @”を含むツイートを除外します.
ちなみにただのリツイートはホームタイムライン取得のレスポンスには入ってきません.

for status in timeline:
    # リツイートを除外
    if not "RT @" in status.text[0:4]:
        if "media" in status.entities:
            for media in status.extended_entities["media"]:
                # Media URL取得
                media_url = media["media_url"]

引用リツイート

元ツイートのURLをTweet本文に貼るだけで引用リツイートになります.
そしてアプリでリツイートした時と同様に, きちんと引用されたユーザに通知されます.