競技プログラミングの日程を通知する 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 の情報を取得して、スクレイピングをします。
その後必要なデータのみを取り出して整形しています。

utils.py

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 を叩いてそれを整形しています。

utils.py

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 というパッケージを使用しています。

db.py

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 テンプレートをストックしておき、それを利用してメッセージを送信します。

main.py

@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 に全てのソースコードが乗っています。
ソースコード