Go to Qiita Advent Calendar 2024 Top
LoginSignup
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

@BQZet

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

Last updated at Posted at 2022-01-29

あらすじ

前回，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が表示されてたら繋がってることがわかる。

image.png

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

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

image.png

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

犬の画像をあげてみる

image.png

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

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

image.png

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

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

おわりに

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

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

では！！！！

1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?