LINE以外のアプリの通知を切っていますか？ 僕は切っています。
AtCoderコンテストの予定が追加されたらLINEで通知してほしいなと思い、Mr JJ と共に LINE BOTを作成しました。
作成したもの
毎時00分に予定されたコンテストの検索を行い、新たなコンテストが追加されている場合BOT君が通知してくれます。
一応メッセージを送ると返してくれる機能もあります。(おまけ)
作成したLINE BOTは以下のボタンから友達追加できます。
バックエンドの構成
バックエンドの処理はすべてAWS(Amazon Web Services)上で行っています。
AWS上の動作は以下の2つに分割できます。
-
コンテスト通知セクション
- AtCoderコンテストの検索と新たなコンテストの通知を行う
-
LINE応答セクション
- BOTに対しメッセージが送信された際に応答を行う
各セクションについて実装した機能を説明します。
コンテスト通知セクション
コンテスト通知セクションの流れ
- Lambda関数: AtCoder_contest_Search がAtCoderホームページのスクレイピングを行い、予定されたコンテスト一覧を取得する
- AtCoder_contest_Search が S3 から通知済みコンテスト一覧 JSON ファイルを取得する。
- 差集合を計算し、未通知コンテストを抽出する。未通知コンテストが空なら終了。
- 未通知コンテストデータをLambda関数: LINE_contest_notify に送る。
- LINE_contest_notify が送られたデータを line-bot-sdk を用いてブロードキャスト形式で通知する。
- [1] で得られたコンテストデータを S3 にアップロード
スクレイピング機能
def get_contests():
# URLからデータを取ってくる
url = "https://atcoder.jp/home?lang=ja"
html_data = requests.get(url)
# htmlパース
soup = BeautifulSoup(html_data.text, "html.parser")
tags = ["upcoming", "recent"]
ret_dic = {}
for tag in tags:
div = soup.find("div", id="contest-table-"+tag)
table = div.find("table")
times = table.find_all("time")
titles = table.find_all("a")
ret_list = []
for i in range(len(times)):
dic = {}
dic["start"] = times[i].text
dic["title"] = titles[i*2+1].text
dic["url"] = "https://atcoder.jp" + titles[i*2+1].get("href")
ret_list.append(dic)
ret_dic[tag] = ret_list
return ret_dic
| データ内容 | |
|---|---|
| 引数 | null |
| 返り値 | 各コンテストに対してデータ(開始時刻・タイトル・URL)を要素に持つ辞書のリスト |
本BOTでは ret_dic["upcoming"] のみ使用。
S3からのファイル取得
def get_data_s3():
# S3上にある通知済みコンテストjsonファイル"contests.json"の読み込み
client = boto3.client("s3")
try:
response = client.get_object(Bucket = "bucket_name", Key = "json_file_name")
except Exception as e:
print(e)
return json.loads(response["Body"].read().decode())
| データ内容 | |
|---|---|
| 引数 | null |
| 返り値 | S3上のJSONファイルを読み込んだ辞書データ |
get_object メソッドの返り値は辞書であり、キー"Body"にファイルの内容が含まれる。
S3へのファイルアップロード
def put_data_s3(save_dic):
# 新たな通知済みコンテストjsonファイルのS3アップローダー
text = json.dumps(save_dic, ensure_ascii = False)
s3 = boto3.resource("s3")
s3.Object("bucket_name", "json_file_name").put(Body = text)
| データ内容 | |
|---|---|
| 引数 | 保存対象の辞書データ |
| 返り値 | null |
辞書をJSON形式のテキストにする際、ensure_ascii の値をFalseにしておくと、
JSONファイル内の日本語が読みやすくなり、ファイルサイズも小さくできる。
ブロードキャスト形式のメッセージ送信
def DatetimeToString(date):
if not isinstance(date, datetime.datetime):
date = datetime.datetime.strptime(date, "%Y-%m-%d %H:%M:%S%z")
week_list = "月火水木金土日"
return f"{date.strftime('%m/%d')}({week_list[date.weekday()]}){date.strftime('%H:%M')}"
def MakeMessageText(event):
text = DatetimeToString(event["start"]) + "\n" + event["title"] + "が開催されます！\n" + event["url"]
return text
def main(event, context):
line_bot_api = LineBotApi("channel_access_token")
message = MakeMessageText(event)
line_bot_api.broadcast(TextSendMessage(text=message))
# *** main関数を終える処理 ***
DatetimeToString 関数でコンテスト開始時間のメッセージフォーマットを作成し、
MakeMessageText 関数で通知するメッセージのフォーマットを作成した。
line-bot-sdk の LineBotApi クラス broadcast メソッドを呼び出すだけでBOTの友達全員に通知を行うことができる。
LINE応答セクション
本システムの構成図
LINE応答セクションの流れ
- LINE BOT がLINEメッセージを受け取るとWebhookを利用してAPI Gatewayへメッセージ内容をPOSTする。
- API GatewayへのPOSTがトリガーとなりLambda関数: LINE_reply_message が実行される。
- LINE_reply_message が line-bot-sdk を用いてLINEのリプライメッセージを送信する。
リプライメッセージ送信
def main(event, context):
handler = WebhookHandler("channel_secret")
line_bot_api = LineBotApi("channel_access_token")
body = event["body"]
# handlerにリプライ関数を追加
@handler.add(MessageEvent)
def SendReply(line_event):
message = MakeReplyMessage(line_event.message.text)
line_bot_api.reply_message(line_event.reply_token, TextSendMessage(text=message))
try:
handler.handle(body, event["headers"]["X-Line-Signature"])
except # エラー処理
# *** main関数を終える処理 ***
MakeReplyMessage 関数はBOTが受け取ったメッセージ内容に応じたリプライメッセージを文字列で返す。
ここの部分に関しては特に line-bot-sdkのドキュメント を読むことをおすすめします。
終わりに
作成した主な機能について紹介を行いました。
簡単にBOTの操作を行えるline-bot-sdk がとても便利です。
AWSに関する知識が少なく、かなり荒い設計になりましたが、気が向いたら今後機能追加＆改善します。(たぶん)