この記事は、AWS LambdaとServerless Advent Calendar 2020の11日目の記事です
昨日の記事は_kenshさんによる「AWS Lambda - Provisioned Concurrencyのベストプラクティス」でした
こんにちは、meiheiです
Clash Royaleというゲームでクランの運営をしています
このゲームではAPI（Clash Royale API）が提供されており、ゲーム内のデータを取得することが出来ます
このAPIを使って、１週間以上の放置しているクランメンバーを取得し、LINEにアラートを送るシステムを作っていきます
概要
完成したもの
こんな感じに、プレイヤー名と放置している期間がLINEに届きます
構成
今回作っていくシステムの構成は以下の通りです
クラロワAPIはIPアドレスを指定する必要があるので、NAT GatewayにEIPを割り当てています
- VPCの設定
- Clash Royale APIのKeyの作成
- Line Notifyの設定
- Lambda関数の作成・デプロイ（Serverless Framework）
の順で説明していきます
VPCの設定
以下の記事を参考（というか全く同じ）に設定していきます
まず、VPCを作成します
適当な名前タグを設定して、IPv4 CIDR ブロックは10.0.0.0/16にします
次に適当な名前のインターネットゲートウェイを作成して、先程作ったVPCにアタッチします
そしてVPCの中に、publicとprivateの２つのサブネットを作成します
「サブネットを作成」からVPCにvpc-lambdaを選択
適当な名前のサブネット名を設定して、IPv4 CIDR ブロックは10.0.10.0/24にします
同様にprivateのサブネットも設定します。IPv4 CIDR ブロックは10.0.1.0/24にします
次にVPC作成時に作られたルートテーブルを選択して、送信先に0.0.0.0/0、ターゲットに先ほど作成したインターネットゲートウェイを設定します
次に、IPアドレスを固定するNAT Gateway＆EIPを作成します（VPCの設定はあとちょっとです！）
適当な名前を入力し、サブネットに先程作ったpublicの方のサブネットを選択します。そしてElastic IP 割り当てというボタンを押してEIPを生成し、NAT Gatewayを作成します。
最後に、VPCの新規ルートテーブルを作成し、送信先0.0.0.0/0、ターゲットに先ほど作成したNAT Gatewayを設定します
サブネット関連付けの編集からprivateのサブネットを関連付けで完了です
これで、Elastic IPの割り当てられた IPv4 アドレスに表示されているIPアドレスが、LambdaからAPIを叩く時に使われる固定IPアドレスになります
クラロワAPIのKeyの作成
前提として、クラロワのdeveloperアカウントを作成してあるとします。
developer.clashroyale.com の右上の名前のところからMy Accountを選択し、Create New Keyから新しい鍵を作成します
KEY NAMEにわかりやすい名前、DESCRIPTIONにわかりやすい説明文、ALLOWED IP ADDRESSに先程作成したIPアドレスを入力し、Create Keyで作成します。
するとKEY NAMEで設定した名前の鍵が生成されています
こちらのTOKENを使います
LINE Notifyの設定
LINE Notify はWEBサービスから簡単にLINEに通知を送れるサービスです
まずログインしてもらって、右上の名前のところからマイページに移動し、下の方にあるアクセストークンの発行(開発者向け)からトークンを発行するを選択します
適当なトークン名を記入し、1:1でLINE Notifyから通知を受け取るを選択してトークンを発行します
発行したトークンが表示されるので、こちらをコピーして保存します（トークンが表示されるのはコレきりなので、コピーを忘れたら再発行しましょう）
Lambda関数の作成・デプロイ（Serverless Framework）
前提として、Serverless Frameworkが使える環境にします。以下の記事がおすすめです
新規のサービスを作成します
$ serverless create --template aws-python3 -p cr-abandon-to-line
次に、必要なプラグインを入れます
$ npm install --save serverless-python-requirements
$ npm install -D serverless-dotenv-plugin
serverless-python-requirementsはpip installした外部モジュールを取り込みます（参考）
serverless-dotenv-pluginはserverless.ymlに直接書きたくない環境変数を設定ファイルからいい感じ取り出してくれます（参考）
プロジェクトディレクトリ下に.envファイルを作成し、先程発行したトークンなどを入れておきます
CR_ACCESS_KEY=[クラロワAPIのトークン]
LINE_NOTIFY_ACCESS_TOKEN=[LINE Notify用に発行したトークン]
VPC_SECURITY_GROUP_IDS=[VPCのセキュリティグループのID]
VPC_SUBNET_IDS=[private-subnetのid]
LambdaがVPCのprivateサブネットに入るようにserverless.ymlに記述します（参考）
iamRoleStatements:
- Effect: "Allow"
Action:
- "ec2:CreateNetworkInterface"
- "ec2:DescribeNetworkInterfaces"
- "ec2:DeleteNetworkInterface"
Resource:
- "*"
vpc:
securityGroupIds:
${env:VPC_SECURITY_GROUP_IDS}
subnetIds:
${env:VPC_SUBNET_IDS}
また、Pythonから環境変数として呼び出しを行うので、その設定もserverless.ymlに記述します
environment:
CR_ACCESS_KEY: ${env:CR_ACCESS_KEY}
LINE_NOTIFY_ACCESS_TOKEN: ${env:LINE_NOTIFY_ACCESS_TOKEN}
そして、クラロワAPIを叩いて、データを整形し、LINE Notifyに送るように実装します。
import json
import os
import requests
from datetime import datetime, timedelta
from typing import Dict, List
CR_BASE_URL = 'https://api.clashroyale.com/v1'
LINE_NOTIFY_URL = "https://notify-api.line.me/api/notify"
CR_ACCESS_KEY = os.environ['CR_ACCESS_KEY']
LINE_NOTIFY_ACCESS_TOKEN = os.environ['LINE_NOTIFY_ACCESS_TOKEN']
def init_headers(api_key: str) -> Dict[str, str]:
"""初期化されたヘッダー情報の辞書を返す"""
return {'authorization': f'Bearer {api_key}'}
def get_member(clan_tag: str) -> Dict[str, str]:
"""クラロワAPIからclan_tagのメンバー情報をGETする"""
clan_tag = clan_tag.replace('#', '%23')
url = CR_BASE_URL + f'/clans/{clan_tag}/members'
headers = init_headers(CR_ACCESS_KEY)
res = requests.get(url, headers=headers)
return res.json()
def last_seen_to_datetime(last_seen: str) -> datetime:
"""last_seenの記法からdatetime型に変換する"""
return datetime.strptime(last_seen, '%Y%m%dT%H%M%S.000Z')
def filter_by_last_seen(items: List[dict], dead_line: datetime) -> List[dict]:
"""最終ログインがdead_lineのクラメンの情報を抽出する"""
before_last_seen = lambda i: last_seen_to_datetime(i['lastSeen']) < dead_line
return [item for item in items if before_last_seen(item)]
def generate_message(cr_items: List[dict]) -> str:
"""line notify送信用のメッセージを作成する"""
message = ''
for member in cr_items:
last_seen_diff = str(datetime.now() - last_seen_to_datetime(member['lastSeen'])).split('.')[0]
message += f'\n{member["name"]}: {last_seen_diff}'
return message
def send_line(message: str):
"""LINE Notifyにmessageを送る"""
headers = init_headers(LINE_NOTIFY_ACCESS_TOKEN)
data = {'message': message}
requests.post(LINE_NOTIFY_URL, data=data, headers=headers)
def lambda_function(event, context) -> Dict[str, str]:
"""実行用の関数"""
data = get_member('#228UCY92')
dead_line = datetime.now() - timedelta(days=1)
filtered_data = filter_by_last_seen(data['items'], dead_line)
if (filtered_data):
message = generate_message(filtered_data)
send_line(message)
return f'ok: {len(filtered_data)}件取得しました'
最後に、hundler.pyのlambda_functionが定期的に実行されるようにserverless.ymlに記述します（参考）
functions:
abandon-member-to-line-notify:
handler: handler.lambda_function
events:
- schedule: cron(0 0 * * ? *) # 日本時間で朝の９時
AWSにデプロイしたら、このシステムの完成です
$ serverless deploy
終わりに
何でも楽にやってくれるLambdaさんだと思っていたら、IPアドレスが固定できないため、なかなか面倒くさい構成になりました
気になるお値段は11日ぐらい使って1,500円ぐらい。月換算だと4,000円ぐらいです（高い）
NAT Gatewayが動いているだけでお金がかかるので、こんなにも高くついているようです
AWSの勉強目的だといいですけど、クラロワAPI＋Lambdaは普通にオススメできないですね。
ソースコード
今回のソースコードはすべてGitHubにあげてあります。READMEは書いてません…
クランメンバー募集
僕の運営しているクラン「ことりカフェ」ではメンバーを募集しています
ほぼ毎日クラロワやってるけど、そこまでガチ勢じゃない、まったりプレイヤーが集うクランです
興味ある方はぜひTwitterにてご連絡ください！
明日は22tさんによる「chaliceで簡単pythonボット制作」です。お楽しみに！