以前ディズニー待ち時間アプリをPythonで再現してみましたが、来月ディズニーに行くことになったので、急遽ラインから呼び出せるようにBotを作成してみることにしました。前回と違って、サーバーでコードを動かし続けてるので誰でも24時間利用することができます。
利用方法としては、アトラクションの頭文字１文字(ひらがなもしくはカタカナ）を入れると、そのアトラクションの待ち時間が返ってくる仕組みになっています。アトラクションの頭文字が26文字の中で良い感じに分散してることと、ピンポイントで目的のアトラクションへリーチできることを考えると、意外と待ち時間アプリよりも使い勝手が良い(?)かなと思います(起動時間は公式よりも短かった)
Botが反応するのは一文字限定なので、Botをグループに入れても、問題なくチャットすることができます。是非ともディズニーに行くときは活用してみてください。
一文字と言いましたが、実はオタク用の隠し要素がちょくちょくあるので、探してみてください(該当コードは載せてません）。
ここからは技術的な解説です。今回のラインbotはBeautifulSoup(スクレイピング)とLine Messaging Apiを活用します。まずGithubのLine messaging api for SDKからコードを引っ張ります。
from flask import Flask, request, abort
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.exceptions import (
InvalidSignatureError
)
from linebot.models import (
MessageEvent, TextMessage, TextSendMessage,
)
app = Flask(__name__)
line_bot_api = LineBotApi('YOUR_CHANNEL_ACCESS_TOKEN')
handler = WebhookHandler('YOUR_CHANNEL_SECRET')
@app.route("/callback", methods=['POST'])
def callback():
# get X-Line-Signature header value
signature = request.headers['X-Line-Signature']
# get request body as text
body = request.get_data(as_text=True)
app.logger.info("Request body: " + body)
# handle webhook body
try:
handler.handle(body, signature)
except InvalidSignatureError:
print("Invalid signature. Please check your channel access token/channel secret.")
abort(400)
return 'OK'
@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text=event.message.text))
if __name__ == "__main__":
app.run()
Line Messagin APIのWebページからチャンネルアクセストークンとチャンネルシークレットを取得してください。Line Messaging APIの細かい説明は省きますが、重要なのは一番下のhandle_message関数です。
event.message.textがチャットで打った文字列が格納されており、現状ではオウム返しをするコードになっています。これを下記のように書き換えます。
@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
if len(event.message.text) == 1:
previous_time = datetime.datetime.now()
word = event.message.text
line = waittime_sea(word)
interval = datetime.timedelta(seconds=1)
current_time = datetime.datetime.now()
if previous_time + interval > current_time:
ar_time = datetime.datetime.now()
line = waittime_sea(word)
inte = datetime.timedelta(seconds=1)
nw_time = datetime.datetime.now()
if ar_time + inte > nw_time:
line = waittime_sea(word)
line_bot_api.reply_message( event.reply_token, TextSendMessage(text=line))
これはLINEで打ち込んだ文字列が1文字か確認して、1文字ならwaittime_sea関数を実行するようになっています。返り値をlineに格納し、最後のTextSendMessageでユーザーに送り返します。階層が深くなってますが、これはスクレイピングできなかった時の対応策です。エラーハンドリングしてもよかったのですが、スクレイピング間隔を調整するためにdatetimeモジュールで判断しています(1秒以上あけて３回までアクセス)。本当はscheduleモジュール、もしくはherokuの定期実行で5分間隔くらいでデータベースに入れたかったのですが、色々と面倒くさかったので諦めました。
def waittime_sea(word):
word = jaconv.hira2kata(word)
url = 'https://tokyodisneyresort.info/realtime.php?park=sea'
res = requests.get(url)
soup = BeautifulSoup(res.text, "html.parser")
attraction = []
wait_time = []
for attraction_temp in soup.find_all(class_="realtime-attr-name"):
attraction.append(attraction_temp.text.strip())
for wait_time_temp in soup.find_all(class_="realtime-attr-condition"):
wait_time_treat = wait_time_temp.text.split("分")[0].strip()
if wait_time_treat.isdecimal():
wait_time_treat += "分"
if "案内終了" in wait_time_treat:
wait_time_treat = '案内終了'
wait_time.append(wait_time_treat)
temp_list = list(zip(attraction, wait_time))
s = ''
for i, _ in enumerate(temp_list):
tup1 = (temp_list[i][0], temp_list[i][1])
line = ':'.join(tup1)
line += '\n'
if line.startswith(word):
s += line
return s
コアとなる関数です。まずラインで打ち込んだ文字列が、ひらがなの場合に自動的にカタカナに変更されるライブラリを活用します。jacovというライブラリがその役割を担ってくれているようなので、事前にpipでインストールします。その後はBeautifulSoupでスクレイピングして、アトラクション名と待ち時間をそれぞれのリストに格納します。
そしてそれぞれのリストをタプルにして取り出し、改行して連結しています。その中で、打ち込んだ文字列で始まるアトラクションだけを抽出します。案内終了の時だけ、うまくテキストが処理できなかったので、改めてリストに手動で「案内終了」と入れ直しました。
これでHerokuにアップロードすると、24時間で稼働するようになります。
こんな感じ！（閉園以降だと少し文字がズレるかな）
果たして実用性がどこまであるか疑問ですが、次回行くときに実際にグループチャットに入れてやってみます。
そもそも入場規制で今は空いてるから、あんまり待ち時間関係ないか。。。。
