競技プログラミングの日程を通知する Line Bot をアップデートしました。
DB 等も使ったので少しまとめていこうと思います。
作ったもの
コンテストという文字を含んだ文章を Bot 本体または Bot が入っているグループに送ると、
Codeforces や AtCoder のコンテスト日程を送ってくれます。
ちなみに下記の QR コードから友達登録ができます。
ぜひ使ってください。
使ったもの
Python
メイン言語として使用PostgreSQL
データベースとして使用ngrok
ローカル環境での Bot のテストに使用heroku
本番環境に使用
アドオンは PostgreSQL と Scheduler を利用しています。Line Bot SDK
Line 公式から提供されている Bot 開発キットです
仕組み
AtCoder のコンテスト日程の取得
AtCoder には公式のコンテスト日程の API が存在しないため、
公式のページをスクレイピングして Rated なコンテストの日程を取ってきています。
スクレイピングには Python の BeautifulSoup を使用しています。
はじめに URL の情報を取得して、スクレイピングをします。
その後必要なデータのみを取り出して整形しています。
def get_upcoming_at_contests():
r = get_data(AT_URL, True)
soup = BeautifulSoup(r, 'html.parser')
texts = soup.get_text()
words = [line.strip() for line in texts.splitlines()]
upcoming = False
text = []
for word in words:
if word == '◉' or word == '':
continue
if word == 'Upcoming Contests':
upcoming = True
continue
if word == 'Recent Contests':
upcoming = False
if upcoming:
text.append(word)
res = []
for i in range(len(text)):
if i < 4:
continue
if i % 4 == 0:
text[i], text[i + 1] = text[i + 1], text[i]
if i % 4 == 1:
s = ''
if i == 1:
pass
else:
for t in text[i]:
if t == '+':
break
s += t
start = datetime.datetime.strptime(s, '%Y-%m-%d %H:%M:%S')
dur = datetime.datetime.strptime(text[i + 1], '%H:%M')
end = start + datetime.timedelta(hours=int(dur.strftime('%H')), minutes=int(dur.strftime('%M')))
s += ' - '
s += end.strftime('%Y-%m-%d %H:%M:%S')
text[i] = s
if i % 4 != 2:
res.append(text[i])
return res
Codeforces の日程取得
Codeforces には公式 API が存在するため、API を叩いてそれを整形しています。
def get_upcoming_cf_contests():
JST = datetime.timezone(datetime.timedelta(hours=+9), 'JST')
contents = get_data(CF_URL)
if contents['status'] == 'FAILED':
print('Failed to call CF API')
return
res = []
for i in range(len(contents['result'])):
if (contents['result'][i]['phase'] == 'FINISHED'):
break
res.insert(0, contents['result'][i]['name'])
start = contents['result'][i]['startTimeSeconds']
s = ''
start_jst = datetime.datetime.fromtimestamp(start, JST)
start_time = datetime.datetime.strftime(start_jst, '%Y-%m-%d %H:%M:%S')
s += start_time
dur_sec = contents['result'][i]['durationSeconds']
dur = datetime.timedelta(seconds=dur_sec)
end_time = start_jst + dur
s += ' - '
s += end_time.strftime('%Y-%m-%d %H:%M:%S')
res.insert(1, s)
return res
DB
heroku にて公式アドオンが提供されている PostgreSQL を使用しています。
レコードの挿入や取得の際は、あらかじめ関数で SQL 文を生成し、それを実行させています。
Psycopg2 というパッケージを使用しています。
import psycopg2
def update_at_table():
query = ''
query += 'DELETE FROM {};'.format(AT_TABLE)
data = utils.format_at_info()
for i in range(len(data)):
query += 'INSERT INTO {0} (name, time, range) VALUES (\'{1}\', \'{2}\', \'{3}\');'.format(AT_TABLE, data[i]['name'], data[i]['time'], data[i]['range'])
execute(query)
def update_cf_table():
query = ''
query += 'DELETE FROM {};'.format(CF_TABLE)
data = utils.format_cf_info()
for i in range(len(data)):
query += 'INSERT INTO {0} (name, time) VALUES (\'{1}\', \'{2}\');'.format(CF_TABLE, data[i]['name'], data[i]['time'])
execute(query)
def get_records(table_name, range=True):
query = ''
if range:
query += 'SELECT name, time, range FROM {};'.format(table_name)
else:
query += 'SELECT name, time FROM {};'.format(table_name)
res = execute(query, False)
return res
def execute(query, Insert=True):
with get_connection() as conn:
if Insert:
with conn.cursor() as cur:
cur.execute(query)
conn.commit()
else:
with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
cur.execute(query)
res = cur.fetchall()
return res
また、heroku のスケジューラで 1 時間に 1 回 DB の内容をアップデートしています。
Line へのデータの送信
公式の SDK で提供されているように、Flask を使用して Web アプリケーションを作成しました。
アプリケーションが起動するとはじめに callback 関数が呼ばれます。
callback 関数内で handle_message 関数が呼ばれメッセージを送信します。
メッセージ送信には Flex Message を利用しています。
Flex Message は Json 形式で生成するため、あらかじめ Json テンプレートをストックしておき、それを利用してメッセージを送信します。
@app.route("/callback", methods=['POST'])
def callback():
signature = request.headers['X-Line-Signature']
body = request.get_data(as_text=True)
app.logger.info('Request body: ' + 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):
user_id = event.source.user_id
to = user_id
if hasattr(event.source, "group_id"):
to = event.source.group_id
TARGET = 'コンテスト'
if not TARGET in event.message.text:
return
cf_data = utils.send_cf_info()
cf_message = FlexSendMessage(
alt_text='hello',
contents=cf_data
)
at_data = utils.send_at_info()
at_message = FlexSendMessage(
alt_text='hello',
contents=at_data
)
try:
line_bot_api.push_message(
to,
messages=cf_message)
line_bot_api.push_message(
to,
messages=at_message)
except LineBotApiError as e:
print('Failed to Send Contests Information')
heroku へのデプロイ
はじめにプロジェクト内に Procfile を作成します。
これは heroku にプロジェクトのタイプや動かし方を伝えるものです。
以下を記述します。
web: python /app/src/main.py
heroku にプロセスタイプを Web であること、また実際に動かすプログラムを教えています。
また、デプロイには heroku CLI を使用しました。
これにより
$ git push heroku master
で heroku へのデプロイができるようになります。
終わり
何か疑問点やわからないこと、改善要求や間違っている点などありましたら Twitter などからぜひ教えてください。
また、GitHub に全てのソースコードが乗っています。
ソースコード