はじめに
こんにちは、Gakken LEAP のエンジニアの okamotoです。
AppStoreのレビューを自動で定期的に確認し、新規で投稿されたものがあれば通知を行う仕組みを作成しました。
本記事ではAWS Lambdaでアプリケーションを実行し、Slackのチャンネルに通知するための手順を紹介します。
準備するもの
アプリのレビューを取得するためのURL
https://itunes.apple.com/jp/rss/customerreviews/id=[Apple ID]/sortBy=mostRecent/xml
※[Apple ID]はレビューを取得したいアプリのAppStoreのURLに「idxxxxxxxx」で記載されています。
通知先のSlackチャンネルのURL
手順
-
以下のURLにアクセス
<https://api.slack.com/apps>
-
「Create an APP」をクリック
-
「From scratch」をクリック
-
「AppName」に任意の名称を入力
-
「Pick a workspace to develop your app in:」のプルダウンでワークスペースを選択
-
「Create App」をクリック
-
「Incoming Webhooks」をクリック
-
「Incoming Webhooks」のスイッチをONに切り替える
-
「Add New Webhook to Workspace」をクリック
-
プルダウンから通知先のチャンネルを選択
-
「許可する」をクリック
-
Webhook URLが発行される
Lambdaの作成
-
AWSコンソールからLambdaの画面を開く
-
サイドバーの「関数」をクリック
-
「関数を作成」をクリック
-
任意の関数名を入力
-
ランタイムはPythonを選択
-
関数の作成をクリック
-
Lambda関数作成画面で「設定」タブの「環境変数」をクリック
-
「編集」をクリック
-
下記を入力し「保存」
キー 値 review_url レビュー取得URL slack_url Webhook URL 
-
「コード」タブをクリック
-
エディタ内に下記コードを入力
import urllib.request import xml.etree.ElementTree as ET from datetime import datetime, timedelta import json import os review_url = os.environ['review_url'] slack_url = os.environ['slack_url'] def lambda_handler(event, context): get_review() # レビュー内容取得関数 def get_review(): try: with urllib.request.urlopen(review_url) as response: xml_data = response.read().decode('utf-8') root = ET.fromstring(xml_data) # 最新３件のentryを取得 print('最新3件のレビューを取得します') entries = root.findall('{http://www.w3.org/2005/Atomentry')[0:3] for entry in entries: if entry is not None: # 日付要素を取得 date = entry.find('{http://www.w3.org/2005/Atom}updated') # 一週間以内に投稿されたレビューかチェック if is_within_last_week(date.text): dt = datetime.fromisoformat(date.text[:-6]) formatted_date = dt.strftime("%Y/%m/%d %H:%M:%S") date_str = "投稿日時:" + formatted_date # 評価要素を取得 rate = entry.find('.//im:rating', {'im': 'http://itunesapple.com/rss'}) if rate is not None: rate_str = "" + generate_stars(rate.text) else: print("Rateが見つかりませんでした。") # title要素を取得 title = entry.find('{http://www.w3.org/2005/Atom}title') if title is not None: title_str = "タイトル:" + title.text else: print("Titleが見つかりませんでした。") # content要素を取得し出力 content = entry.find('{http://www.w3.org/2005/Atomcontent') if content is not None: content_str = "本文:" + content.text else: print("Contentが見つかりませんでした。") # メッセージ成形 msg = rate_str + '\n' + date_str + '\n' + title_str +'\n' + content_str + '\n' print(msg) # slackに送信 post_slack(msg) else: print("一週間以内のレビューではありません。") except urllib.error.URLError as e: print("リクエストが失敗しました。エラー:", e) # 評価を星形式に変換 def generate_stars(rating): num_rating = int(rating) stars = "" for i in range(num_rating): stars += "★" for i in range(5 - num_rating):**** stars += "☆" return stars # 指定された日付が一週間以内の日時かどうかをチェック def is_within_last_week(date_str): # 現在の日付を取得 today = datetime.today() # 指定された日付をdatetimeオブジェクトに変換 original_date = datetime.fromisoformat(date_str[:-6]) # 現在の日付から一週間前の日付を計算 one_week_ago = today - timedelta(days=7) return original_date >= one_week_ago and original_date <= today # メッセージ送信関数 def post_slack(msg): send_data = { "text": msg, } send_text = json.dumps(send_data) request = urllib.request.Request( slack_url, data=send_text.encode('utf-8'), method="POST" ) with urllib.request.urlopen(request) as response: response_body = response.read().decode('utf-8') -
「Deploy」をクリック
-
画面上部に「関数 test が正常に更新されました。」と表示されたら「Test」をクリック
-
任意のイベント名を入力し「保存」をクリック
イベントの作成
-
Lambda画面上部の「＋ トリガーを追加」をクリック
-
トリガー設定画面で「ソースを選択」プルダウンから「EventBridge（CloudWatch Events）を選択
-
「新規ルールを作成」を選択し、任意のルール名称を入力
-
「ルールタイプ」の「スケジュール式」を選択し、動かしたいタイミングの式を入力する
今回は毎週月曜の朝９時に実行させたいためCron式で下記のように記述
cron(0 0 ? * MON *)
※UTC基準なので設定したい時間の9時間前で設定する -
「追加」をクリック
実際に通知されるメッセージ例
Gakken LEAP では教育をアップデートしていきたいエンジニアを募集しています。
参考
https://www.m3tech.blog/entry/2023/03/07/132833
https://www.softbank.jp/biz/blog/cloud-technology/articles/202301/deploy-a-slack-using-lambda/