【python】TwitterのAPI（v2）を使ってリプライに添付された画像に対して返信するbotを作ってみた(2/2 記述したソースコードをAWSLambda定期実行でbotを稼働させる)

あらすじ

前回，TwitterAPIv2を用いて検索・画像URL取得・ツイート送信・リプライ送信まで説明したので，今回は実際にbotを作っていきます！

実行環境

• python3.9.2
• windows11
• TwitterAPI v2
• foodAI v4.1

実装方針

作成するクラス・関数

大きく分けて，2つのクラスと1つの関数で実装する。

• Twitterapiクラス - 引数なし
  • searchメソッド - 引数に検索クエリを与えると検索結果を返す(/2/tweets/search/recent)
  • replyメソッド - 引数にリプライ先のツイードIDとリプライメッセージを与えると，リプライを送信できる(/2/tweets)
• model.Tweetクラス - 引数なし。dynamodbのテーブルとやり取りを行うクラス。検索で得られたツイートが処理済みかどうか判別するために使う
  • insertメソッド - 引数にツイートのjsonを入れるとテーブルに追加される
  • selectメソッド - 引数にツイードIDを入れると，そのツイートがテーブルに存在している場合，ツイートを返してくれる
• Food_identifyクラス - 引数に画像URLを張るとfoodAIに投げて結果を返す
  • judgeメソッド - 結果が食品かそうでないか判別する
• lambda_handler - lambdaで実行する関数

TwitterapiクラスとFood_identifyクラスを用いてlambda_handlerにbotで用いる処理を記述していく感じ

ディレクトリ構成

app/
　├ lambda_function.py
　└ model.py

TwitterAPIクラス

lambda_function.py

import requests
from requests_oauthlib import OAuth1Session
class Twitterapi:
    def __init__(self):
        apikey = "ClientID"
        secretkey = "Clientsecret"
        accesstoken = "accessToken"
        accesstokensecret = "accessTokenSecret"
        self.oauth = OAuth1Session(
            apikey,
            client_secret=secretkey,
            resource_owner_key=accesstoken,
            resource_owner_secret=accesstokensecret
        )
    #認証用の関数
    def _bearer_oauth(self, r):
        bearertoken = "bearertoken"
        r.headers["Authorization"] = f"Bearer {bearertoken}"
        r.headers["User-Agent"] = "v2RecentSearchPython"
        return r
    # 検索
    def search(self, query):
        params = {}
        params["query"] = query
        params["expansions"] = "attachments.media_keys,author_id"
        params["media.fields"] = "url"
        params["user.fields"] = "username"
        url = "https://api.twitter.com/2/tweets/search/recent"
        response = requests.get(url, auth=self._bearer_oauth, params=params)
        #ステータスコードが200以外ならエラー処理
        if response.status_code != 200:
            raise Exception(response.status_code, response.text)
        #responseからJSONを取得
        return response.json()
    # ツイート
    def reply(self, to_replied_tweet_id, text):
        url = "https://api.twitter.com/2/tweets"
        params = {
            "text": text,
            "reply":{
                "in_reply_to_tweet_id": str(to_replied_tweet_id)
                }
            }
        response = self.oauth.post(url, json=params)
        #ステータスコードが201以外ならエラー処理
        if response.status_code != 201:
            raise Exception(response.status_code, response.text)

ClientIDとかAccessTokenとかは各自用意したやつを記述
使い方は，以下の通り

twitterapi = Twitterapi()

# ツイートを検索
result = twitterapi.search("検索したい文字列")

# 任意のツイートにリプライ
twitterapi.reply("リプライ先ツイートID", "リプライ内容")

model.Tweetクラス

検索で得られたツイートが処理済みかどうか判別するために，処理済みのツイートはdynamodbのテーブルに挿入することにした。
テーブルと繋ぐためのモデルクラスを以下に示す。
boto3ライブラリは，python用のAWSのSDKで，AWSのリソースをpythonで使うためのライブラリである。
このライブラリを使うためには，AWSCLIをインストールして，aws configureコマンドを打ってAPIキーなどを設定しなければいけないが，その辺は割愛･･･

model.py

import boto3
class Tweet:
    dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1')
    table = dynamodb.Table('テーブル名')
    def insert(self, tweet):
        with self.table.batch_writer() as batch:
            batch.put_item(Item=tweet)
    def select(self, key):
        response = self.table.get_item(Key={"id": key})
        try:
            return response["Item"]
        except:
            return None

AWSコンソール上でdynamodbのテーブルを作成し，作成したテーブルのリージョンとテーブル名を指定すること
あと，プライマリーキーは"id"としてる

使い方の一例を以下に示す

# インスタンス作成
twitterapi = Twitterapi()
tweetmodel = Tweet()

'''
検索で得られたツイートをテーブルに挿入
'''
# 検索処理
tweetlist = twitterapi.search("検索したい文字列")
# テーブルに挿入
for tweet in tweetlist["data"]:
     tweetmodel.insert(tweet)

'''
検索で得られたツイートがすでにテーブルに存在するか判断
'''
# 検索処理2
tweetlist2 = twitterapi.search("検索したい文字列2")
# ツイートが存在するか？
for tweet in tweetlist["data"]:
     if tweetmodel.select(tweet["id"]):
          print(f"ツイートID：{tweet['id']}\n内容：{tweet['text']}\nは既に存在します")
     else:
          print(f"ツイートID：{tweet['id']}は存在しません")

FoodAIクラス

lambda_function.py

class Food_identify:
    def __init__(self, image):
        '''
        return: 一番適応度が高かった食品の名称，適応度
        '''
        url = 'https://api.foodai.org/v4.1/classify'
        payload = {
            'image_url' : image,
            'num_tag' : 10,
            'api_key':'apikey'
        }
        self.res = requests.post(url, data=payload).json()
    # 食べ物だったらTrue, 食べ物じゃなかったらFalseを返す
    def judge(self):
        food = self.res["food_results"][0][0]
        point = float(self.res["food_results"][0][1])
        return food != "Non Food"

"apikey"にはfoodaiのAPIキーを入力
使い方は以下の通り

foodai = Food_identify("画像のURL")
if foodai.judge():
     print("画像は食べ物です")
else:
     print("画像は食べ物ではありません")

lambda_handler関数

lambda_function.py

import model
import random

def lambda_handler(event, context):
    # api認証
    twitterapi = Twitterapi()
    # リプライ検索
    replylist = twitterapi.search("@firstCorgi has:images -is:retweet")
    replylist_data = replylist["data"]
    replylist_media = replylist["includes"]["media"]

    # 既に処理されたツイートを削除
    tweetmodel = model.Tweet()
    for tweet in replylist_data:
        tweet["id"] = int(tweet["id"])
        tweet["author_id"] = int(tweet["author_id"])
        # 検索で得られたツイートがDBに存在しないなら
        if not tweetmodel.select(tweet["id"]):
            # ツイートに添えられた最初の画像のURL取得
            mediakeys = tweet["attachments"]["media_keys"][0]
            for media in replylist_media:
                if media["media_key"] == mediakeys:
                    url = media["url"]
                    break
            # 画像判別
            food_identify = Food_identify(url)
            judge = food_identify.judge()
            # 食品なら"ワンワンワン！！"　食品じゃないなら"グルルルルルルル...."
            if judge:
                 twitterapi.reply(tweet["id"], f"ワン" * int(random.random()*10) + "！" * int(random.random()*10) + "ワン" * int(random.random()*15) + "！" * int(random.random()*15) + "ワン" * int(random.random()*20) + "！" * int(random.random()*20) + "(歓喜)")
            else:
                 twitterapi.reply(tweet["id"], f"グ" + "ル" * int(random.random()*40 + 10)  + "." * int(random.random()*50 + 30) + "(警戒)")
            # DBにツイート挿入
            tweetmodel.insert(tweet)

    return {
        'statusCode': 200,
        'body': "OK"
    }

ソースコードまとめ

app/
　├ lambda_function.py
　└ model.py

lambda_function.py

import model
import random
import requests
from requests_oauthlib import OAuth1Session

class Twitterapi:
    def __init__(self):
        apikey = "ClientID"
        secretkey = "Clientsecret"
        accesstoken = "accessToken"
        accesstokensecret = "accessTokenSecret"
        self.oauth = OAuth1Session(
            apikey,
            client_secret=secretkey,
            resource_owner_key=accesstoken,
            resource_owner_secret=accesstokensecret
        )
    #認証用の関数
    def _bearer_oauth(self, r):
        bearertoken = "bearertoken"
        r.headers["Authorization"] = f"Bearer {bearertoken}"
        r.headers["User-Agent"] = "v2RecentSearchPython"
        return r
    # 検索
    def search(self, query):
        params = {}
        params["query"] = query
        params["expansions"] = "attachments.media_keys,author_id"
        params["media.fields"] = "url"
        params["user.fields"] = "username"
        url = "https://api.twitter.com/2/tweets/search/recent"
        response = requests.get(url, auth=self._bearer_oauth, params=params)
        #ステータスコードが200以外ならエラー処理
        if response.status_code != 200:
            raise Exception(response.status_code, response.text)
        #responseからJSONを取得
        return response.json()
    # ツイート
    def reply(self, to_replied_tweet_id, text):
        url = "https://api.twitter.com/2/tweets"
        params = {
            "text": text,
            "reply":{
                "in_reply_to_tweet_id": str(to_replied_tweet_id)
                }
            }
        response = self.oauth.post(url, json=params)
        #ステータスコードが201以外ならエラー処理
        if response.status_code != 201:
            raise Exception(response.status_code, response.text)

class Food_identify:
    def __init__(self, image):
        '''
        return: 一番適応度が高かった食品の名称，適応度
        '''
        url = 'https://api.foodai.org/v4.1/classify'
        payload = {
            'image_url' : image,
            'num_tag' : 10,
            'api_key':'apikey'
        }
        self.res = requests.post(url, data=payload).json()
    # 食べ物だったらTrue, 食べ物じゃなかったらFalseを返す
    def judge(self):
        food = self.res["food_results"][0][0]
        point = float(self.res["food_results"][0][1])
        return food != "Non Food"

def lambda_handler(event, context):
    # api認証
    twitterapi = Twitterapi()
    # リプライ検索
    replylist = twitterapi.search("@firstCorgi has:images -is:retweet")
    replylist_data = replylist["data"]
    replylist_media = replylist["includes"]["media"]

    # 既に処理されたツイートを削除
    tweetmodel = model.Tweet()
    for tweet in replylist_data:
        tweet["id"] = int(tweet["id"])
        tweet["author_id"] = int(tweet["author_id"])
        # 検索で得られたツイートがDBに存在しないなら
        if not tweetmodel.select(tweet["id"]):
            # ツイートに添えられた最初の画像のURL取得
            mediakeys = tweet["attachments"]["media_keys"][0]
            for media in replylist_media:
                if media["media_key"] == mediakeys:
                    url = media["url"]
                    break
            # 画像判別
            food_identify = Food_identify(url)
            judge = food_identify.judge()
            # 食品なら"ワンワンワン！！"　食品じゃないなら"グルルルルルルル...."
            if judge:
                 twitterapi.reply(tweet["id"], f"ワン" * int(random.random()*10) + "！" * int(random.random()*10) + "ワン" * int(random.random()*15) + "！" * int(random.random()*15) + "ワン" * int(random.random()*20) + "！" * int(random.random()*20) + "(歓喜)")
            else:
                 twitterapi.reply(tweet["id"], f"グ" + "ル" * int(random.random()*40 + 10)  + "." * int(random.random()*50 + 30) + "(警戒)")
            # DBにツイート挿入
            tweetmodel.insert(tweet)

    return {
        'statusCode': 200,
        'body': "OK"
    }

app/
　├ lambda_function.py
　└ model.py

model.py

import boto3
class Tweet:
    dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1')
    table = dynamodb.Table('テーブル名')
    def insert(self, tweet):
        with self.table.batch_writer() as batch:
            batch.put_item(Item=tweet)
    def select(self, key):
        response = self.table.get_item(Key={"id": key})
        try:
            return response["Item"]
        except:
            return None

ソースコードをAWSLambdaにアップロードし，定期実行させる

requests_oauthlibの依存ライブラリをカレントディレクトリに入れる

では，コードもできたし早速zipに圧縮してアップロード･･･
と言いたいところだが，このままではrequests_oauthlibがAWSLambda標準のライブラリじゃないので，実行できない
なので，一時的にvenv環境にrequests_oauthlibをインストールし，依存ライブラリを全てカレントディレクトリに入れる
以下のコマンドを実行

$ python -m venv venv
$ venv\Scripts\activate
$ pip install requests_oauthlib
$ pip install requests -t ./

zip化し，アップロード

カレントディレクトリは以下のようになっていると思う

app/
　├ venv/
　├ bin/
　├ certifi/
　├ certifi-2021.10.8.dist-info/
　├ charset_normalizer/
　├ charset_normalizer-2.0.10.dist-info/
　├ idna/
　├ idna-3.3.dist-info/
　├ requests/
　├ requests-2.27.1.dist-info/
　├ urllib3/
　├ urllib3-1.26.8.dist-info/
　├ lambda_function.py
　└ model.py

appをzip化し，Lambda関数を作成し，zipファイルをアップロードする

EventBridgeを用いてLambda関数を定期実行する

ここまで来たらあと少し！
AWSmanagementConsoleで，EventBridgeを開き，ルールを作成を押下
大体デフォルト設定だが，
パターンを定義のところで，スケジュールを選択し固定時間ごと5分で設定
ターゲットにLambda関数を選択し，機能に作成した関数を選択する

ルール作成を行い，最初はdisabledになってるので，作成したルールを有効にする。

作成したLambda関数に戻り，関数の概要図にEventBridgeが表示されてたら繋がってることがわかる。

作成したbotにリプライを送ってみる

ラーメンの画像をあげてみる

食べ物だから喜んだ！！成功！！

犬の画像をあげてみる

食べ物じゃないから警戒してる！！！

食べ終わったラーメンの画像をあげてみる

もう！！！！！それは喜ぶなよバカ！！！！！！！！！！！！！！！！！！！！

まあ･･･本物のコーギーもバカだからある意味リアルってことでいいかな･･･？？

おわりに

以上，TwitterAPIv2を用いてbotを作成する一連の流れを書きました
雑な文章なので理解できない部分も多いと思いますが，ソースコードはちゃんと動くので参考になると思います

自分で作ったbotは，なんか愛着がでて可愛いですね。
もっと育てて（多機能にして）あげたいです

では！！！！