Apexのランクポイントの増減計算をするLINEBotを作った話

この記事はTDU CPSLab Advent Calendar 2021 - Adventarの2日目の記事です

前の記事はこちらから、次の記事はこちらから。

はじめに

1回ダイヤ上がれたけど、それからずっとプラチナ3くらいまでしか上げられなくて、
最近CoDの方に移動しようか悩んでいるなるせいです。

突然ですが、Apex Legendsやってますか？
日本だとかなり大人気だなって印象を受けます。

好きな配信者さんなどおりますでしょうか？
その方のランク配信とかで隅っこの方に、
「ランクポイントの増減カウンター」みたいなのありませんかね。

今回はそれを対話式に実現するLINEBotを作ってみたので、そのお話をします。

例の如く、車輪の再発明しているかもしれませんが、
何となく作ってみた記事となります。ご容赦ください。

目次

• 開発環境
• どんなことができるの？
• 実装内容
  • オウム返しをしてみよう
  • Apexデータを取得して返してみよう
  • 任意のコマンド経由でやり取りできるようにしよう!
  • dynamoDBを使って情報を保持しておこう
• 検証

開発環境

使っている技術は以下の通りです。

• AWS
  • lambda
  • API Gateway
  • DynamoDB
• LINE Messaging APIを利用できるBot
• Python3.9
  • line-bot-sdk
  • ApexTrackerPy
  • boto3

構成図は以下のような形。

AWSアーキテクチャアイコンを使用しました。

どんなことができるの？

利用手順と機能的には以下の通りです。

1. LINEアカウントを追加する
2. userinfoコマンドを使ってユーザ名とプラットフォームを登録する。
3. 各種コマンドを使ってApexの情報にアクセスできる
   1. rankコマンド
   2. startコマンド
   3. stopコマンド

コマンドの構文と機能説明は表の通りです。

コマンド	構文	機能説明
userinfo	userinfo [username] [platform]	Apexのユーザ名と利用プラットフォームを登録します。プラットフォームにはPC,PS4,X1,SWITCHの中から登録します。
rank	rank	登録されたユーザのランクとランクポイントを表示します
start	start	ランクポイントの変動計算を開始します。コマンド入力時のランクポイントが保持されます
stop	stop	ランクポイントの変動計算を終了します。現在のランクポイントと保持されたランクポイントの差分を表示します

利用イメージとしては、まずuserinfoで登録。startで計測開始してstopで変動値をみようぜ。みたいな感じです。

実装内容

自分はまずLINEBot・lambda関数の作成・API Gatewayの設定を行いました。そこからミニマムにアップロード行い動作確認しながら開発を進めました。記事の内容もそれに倣いたいと思ってます。

また、今回のアプリを作るにあたってはたくさん記事を参考にさせていただきました。従いまして、段階的に全く同じだったりするところがございます。その場合は、そちらの方を御一読くださいと案内させていただきますのご容赦ください。(個人的に)詰まりがちな点は本記事でも明記します。

オウム返しをしてみよう

LINEBot作成の登竜門的な存在である(?)オウム返しBotを作りましょう。
こちらはkonikoni428様のこちらの記事を参考にさせていただきました。
本当にありがとうございます。

line-bot-sdk-pythonを使ったオウム返しBotの作成についてご紹介されております。
丸々参考にさせていただきましたので御一読ください。

一点、LambdaにアップロードするためにZIPにまとめるのですが、それに関しまして公式ドキュメントが分かりやすかったので共有させていただきます。

Apexデータを取得して返してみよう

実装目標に入っていきます。今回Apexデータを取得するにあたっては、nerrog様のこちらの記事を参考にさせていただきました。

参考にというか、今回使用させていただくApexTrackerPyを作られた方です。
本当にありがとうございます。

さて、ここからは簡単にコードを記載します。
その中でも手を加えていた部分というと、オウム返しBotのmessage関数部分になります。
ですので、message関数についてのコードに絞って載せたいと思います。

lambda_function.py

import ApexTrackerPy

apex_api_key = os.getenv('APEX_API_KEY', None)

if apex_api_key is None:
    logger.error('Specify APEX_API_KEY as environment variable.')
    sys.exit(1)

def lambda_handler(event, context):
    # ...
    @handler.add(MessageEvent, message=TextMessage)
    def message(line_event):
        text = line_event.message.text
        res = ApexTrackerPy.GetApexPlayerStatus(apex_api_key, "PS4", "あなたのユーザ名") # ここであなたの戦績を取得している。
        line_bot_api.reply_message(line_event.reply_token, TextSendMessage(text=str(res.CurrentRank))) # 現在のランクを返す

このような感じで書くと何かメッセージを送った際に、あなたのランクをリプライするBotになるかと思います。特筆するべき点としては発行したAPI Keyについてオウム返しBotの際のLINE_CHANNEL_SECRETやLINE_CHANNEL_ACCESS_TOKENと同様にAPEX_API_KEYとしてLambdaの設定の「環境変数」から設定しております。

任意のコマンド経由でやり取りできるようにしよう!

これでApexの情報をLINEBot越しで確認できるようになりましたが、前述したできることの紹介のように今回いろんなコマンド的なものが存在するのでその辺りに関しての処理について記載します。まずコードです。

lambda_function.py

apex_api_key = os.getenv('APEX_API_KEY', None)
command_type = ["userinfo", "rank", "start", "stop"] # 実装するコマンドのリスト

def message(line_event):
    text = line_event.message.text
    text_array = str(text).split()
    command = text_array[0]
    result = ""

    if command in command_type: # 定義されたコマンドが入ってきているか否か
        if command == "userinfo":
            username = text_array[1]
            input_platform = text_array[2]

            platform = ""
            if input_platform.lower() == "pc" or input_platform.lower() == "origin":
                platform = "PC"
            elif input_platform.lower() == "ps4" or input_platform.lower() == "psn":
                platform = "PS4"
            elif input_platform.lower() == "x1" or input_platform.lower() == "xbox":
                platform = "X1"
            elif input_platform.lower() == "switch" or input_platform.lower() == "nintendo":
                platform = "SWITCH"
            # 登録処理は？？？
            result = "ユーザーネームを" + username + "，プラットフォームを" + platform + "で登録しました。"
        elif command == "rank":
            res = ApexTrackerPy.GetApexPlayerStatus(apex_api_key, "PS4", "あなたのユーザ名")
            result = "現在のランク：" + str(res.CurrentRank) + "，ランクポイント：" + str(res.Rank_RP)
        elif command == "start":
            res = ApexTrackerPy.GetApexPlayerStatus(apex_api_key, "PS4", "あなたのユーザ名")
            result = "ランクポイントの増減計測を開始します。現在のランクポイントは" + str(res.Rank_RP) + "です。"
        elif command == "stop":
            res = ApexTrackerPy.GetApexPlayerStatus(apex_api_key, "PS4", "あなたのユーザ名")
            diff = int(res.Rank_RP) - int(過去のランクポイント) #過去のランクポイントどうする？
            result = "ランクポイントの増減計測の結果は、" + str(diff) + "です。現在のランクは" + str(res.CurrentRank) + "です。"
    else:
        result = "申し訳ありません。該当のコマンドを送信してください。ex) userinfo [username] [platform], rank, start, stop"
    line_bot_api.reply_message(line_event.reply_token, TextSendMessage(text=result))

やっている事は、LINEから送られてくるメッセージをsplitで分割してその0番目がコマンドであると捉え、これを条件分岐させることで別々の処理を行なっているだけになります。result変数の中に最終的にレスポンスとして返すメッセージを整形しています。

一方で、ユーザ名やプラットフォームの保持をしていないとコマンドを呼ぶ度に2つの情報が必要になってしまったり、過去のランクポイントの保持をしていないと増減計算などができません。そこでDynamoDBを使用して値を保存しておこうと思います。

dynamoDBを使って情報を保持しておこう

まずLambdaからDynamoDBにアクセスすることができるように設定を変更しておきましょう。
作成した関数の設定を開き、アクセス権限から実行ロールの変更を行います。
画像のように「LambdaAccess2DynamoDB」に変更してください。

これでDynamoDBにアクセスできるようになるかと思います。

追記
「LambdaAccess2DynamoDB」なんてロールは存在しない！！！！
はずです。本当にすみません。

こちらのロールは過去に同様の用途を実現する際に作った独自のロールでした。
更に独自っていっても独自ではなくて、NTT東日本様のコラム記事を参考に作成をしました。

本当にすみません。記事の方をご査収ください。。！
記事の方が詳しいのですが、雑に書いておきますと。

IAMの方でロールというタブからロールの作成を押下します。
AWSサービスの中からLambdaを選択しまして、アクセス権限の設定を行います。
設定するのは記事内では2つでしたが自分は、
「AmazonDynamoDBFullAccess」
「AWSLambdaDynamoDBExecutionRole」
「AWSLambdaBasicExecutionRole」
の3つを設定してます。
その後はロール名をLambdaAccess2DynamoDBに設定していただけばOKです。
(タグとかの他の部分は特に何もせず右下の青いボタンを押していって下さい)
作成した後にアタッチする際には上記の手順で問題ございません。

---

続いてDynamoDBのテーブルを作成しましょう。

検索窓から「DynamoDB」を検索して一番上をクリックします。
次に左側のメニューから「テーブル」を選ぶと右上に「テーブルの作成」というボタンが表示されている画面になるのではないでしょうか？

そのボタンを押下しましたら、テーブル名とパーテションキーを入力します。
今回パーテションキーにはuserIdと指定します。

下部にある「テーブルの作成」ボタンを押下したら作成完了です。

---

続いてコード側です。boto3というPythonにおけるawsのsdkを使用します。
使い方についてはshimajiri様のこちらの記事を参考にさせていただきました。
本当にありがとうございました。

コードは以下のような形です。

lambda_function.py

import boto3
from botocore.exceptions import ClientError

def message(line_event):
    text = line_event.message.text
    text_array = str(text).split()
    command = text_array[0]
    result = ""

    if command in command_type: # 指定コマンド以外ではdynamoDBにアクセスしたくなくてこの条件分岐を入れてみた
        dynamoDB = boto3.resource('dynamodb')
        table = dynamoDB.Table('作成したテーブル名')
        if command == "userinfo":
            username = text_array[1]
            input_platform = text_array[2]

            platform = ""
            # 上記されているので省略。platformの入力値のチェックをする。

            try:
                # 新規作成
                table.put_item(
                    Item = {
                        'userId': line_event.source.user_id,
                        'username': username,
                        'platform': platform,
                        'point_record': "0"
                    }
                )
                result = "ユーザーネームを" + username + "，プラットフォームを" + platform + "で登録しました。"
             except ClientError as e:
                result = "ユーザーネームをとプラットフォームの登録に失敗しました。何度もこのメッセージが表示される場合には管理者にお問い合わせください。"
        elif command == "rank":
            try:
                # 取得
                query_data = table.get_item(
                    Key = {
                        'userId': line_event.source.user_id,
                    }
                )
                res = ApexTrackerPy.GetApexPlayerStatus(apex_api_key, query_data["Item"]["platform"], query_data["Item"]["username"])
                result = "現在のランク：" + str(res.CurrentRank) + "，ランクポイント：" + str(res.Rank_RP)
            except ClientError as e:
                result = "ユーザ情報が取得できません。userinfo [username] [platform] コマンドを入力してください。platformはPC,PS4,X1,SWITCHから入力してください。既に入力できているのに関わらず本メッセージが表示される場合は管理者にお問い合わせください。"
        elif command == "start":
            try:
                # 取得
                query_data = table.get_item(
                    Key = {
                        'userId': line_event.source.user_id,
                    }
                )
                res = ApexTrackerPy.GetApexPlayerStatus(apex_api_key, query_data["Item"]["platform"], query_data["Item"]["username"])

                # 更新
                table.update_item(
                    Key = {'userId': line_event.source.user_id},
                    UpdateExpression = 'set point_record = :s',
                    ExpressionAttributeValues={
                        ':s' : str(res.Rank_RP)
                    }
                )
                result = "ランクポイントの増減計測を開始します。現在のランクポイントは" + str(res.Rank_RP) + "です。"
            except ClientError as e:
                if e.response['Error']['Code'] == 'ResourceNotFoundException':
                    result = "ユーザ情報が取得できません。userinfo [username] [platform] コマンドを入力してください。platformはPC,PS4,X1,SWITCHから入力してください。既に入力できているのに関わらず本メッセージが表示される場合は管理者にお問い合わせください。"
                elif e.response['Error']['Code'] == 'ConditionalCheckFailedException':
                    result = "ランクポイントを更新できません。何度もこのメッセージが表示される場合には管理者にお問い合わせください。"
                else:
                    result = "エラーが発生しました"
        elif command == "stop":
            try:
                # 取得
                query_data = table.get_item(
                    Key = {
                        'userId': line_event.source.user_id,
                    }
                )
                res = ApexTrackerPy.GetApexPlayerStatus(apex_api_key, query_data["Item"]["platform"], query_data["Item"]["username"])

                # 差分の計算。ただ引き算してるだけですね。
                diff = int(res.Rank_RP) - int(query_data["Item"]["point_record"])
                result = "ランクポイントの増減計測の結果は、" + str(diff) + "です。現在のランクは" + str(res.CurrentRank) + "です。"
            except ClientError as e:
                result = "ユーザ情報が取得できません。userinfo [username] [platform] コマンドを入力してください。platformはPC,PS4,X1,SWITCHから入力してください。既に入力できているのに関わらず本メッセージが表示される場合は管理者にお問い合わせください。"
    else:
        result = "申し訳ありません。該当のコマンドを送信してください。ex) userinfo [username] [platform], rank, start, stop"
    line_bot_api.reply_message(line_event.reply_token, TextSendMessage(text=result))

LINEのユーザIDをパーティションキーとするDynamoDBのテーブルに対して、platform, username, point_recordの3要素が1つのデータとして保存、更新される形になります。これでデータを保持し、そのデータをもって増減計算などができるようになりました。

command_typeで条件分岐している箇所はdynamoDBのインスタンスを不要に作りたくないなぁとか思ってこんな処理にしているのですが、boto3への理解が浅く別にこんなことしなくても大丈夫なのだろうかとか今更思ってます。

あとエラー系のところ全部ハードコーディングしちゃってますけど、そういう文字列を返す関数を別に用意してresult = non_find_userinfo_errormessage()みたいにした方が見やすそう。

検証

さてさて。。。正直コードについては特に目立つ工夫はなく普通に書いてるだけだなという感じかと思います。作ってみた記事なので動作確認できるのか検証していきたいと思います。

まずユーザ情報の登録からです。

今回検証するの区間は以下の開始と終了になります。

start 4906	stop 5139

こちらに対してLINEの方は以下のような感じになりました。

5139 - 4906 = 233ですね。ちゃんと動作しているようです！
ちなみに間の73は気になって試しに確認しちゃっただけですので気にしないでください。

DynamoDB方もこのようにちゃんと登録されていました。

まとめ

こういうちょっとしたツールですが、AWSを利用するとサクッと作れてしまうのは感動です。

あと勘がいい方は、薄々気づいているかと思いますが、本Botはアドカレに向けて突貫で作りまして、エラーハンドリングとかちゃんと考慮できていないかと。。。

記事書くのにヒイヒイ言っていたので落ち着いたら見直そうと思います。

それではランク盛れるように頑張りましょう
僕はCoD:Vやります

参考記事

• Lambdaでline-bot-sdk-pythonを使用してオウム返しBOTを作成する
• .zip ファイルアーカイブで Python Lambda 関数をデプロイする
• ApexLegendsの戦歴をPythonで取得してみよう
• AWS LambdaからPython Boto3を使用してDynamoDBを操作してみた
• AWS Python(Boto3)の例外ハンドリングとテスト時のraise方法
• AWS API GatewayとLambdaでDynamoDB操作