[PYTHON] L'histoire de la création d'un Line Bot qui nous raconte le calendrier de la programmation du concours

Mise à jour de Line Bot pour vous informer des dates de programmation du concours. J'ai aussi utilisé DB etc. donc je vais le résumer un peu.

Ce que j'ai fait

Si vous envoyez une phrase contenant le mot "concours" au Bot lui-même ou au groupe qui contient le Bot, Ils vous enverront les dates des concours Codeforces et AtCoder.

En passant, vous pouvez vous inscrire comme ami à partir du code QR ci-dessous. Veuillez l'utiliser. Screen Shot 2020-05-15 at 23.28.15.png

Ce que j'ai utilisé

Python Utilisé comme langue principale
PostgreSQL Utilisé comme base de données
ngrok Utilisé pour tester Bot dans un environnement local
heroku Utilisé pour l'environnement de production L'add-on utilise PostgreSQL et Scheduler.
Line Bot SDK Kit de développement de bot fourni par Line Official

Comment ça fonctionne

Obtenir les dates du concours AtCoder

AtCoder n'a pas d'API de calendrier de concours officiel Je racle la page officielle pour planifier un concours noté. J'utilise BeautifulSoup de Python pour le scraping. Tout d'abord, obtenez les informations d'URL et récupérez-les. Après cela, seules les données nécessaires sont extraites et formatées.

`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

Obtenir les dates Codeforces

Codeforces a une API officielle, alors j'ai frappé l'API pour la formater.

`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 J'utilise PostgreSQL, l'add-on officiel fourni par heroku. Lors de l'insertion ou de la récupération d'un enregistrement, une fonction génère une instruction SQL à l'avance et l'exécute. J'utilise un package appelé 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

De plus, le planificateur heroku met à jour le contenu de la base de données une fois par heure.

Envoi de données à Line

J'ai utilisé Flask pour créer une application Web, comme fourni par le SDK officiel. La fonction de rappel est appelée pour la première fois au démarrage de l'application. La fonction handle_message est appelée dans la fonction de rappel pour envoyer le message. Flex Message est utilisé pour envoyer des messages. Étant donné que Flex Message est généré au format Json, stockez les modèles Json à l'avance et utilisez-les pour envoyer des messages.

`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 = 'Concours' 
    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')

Déployer sur heroku

Tout d'abord, créez un fichier Procfile dans votre projet. Cela indique à heroku de quel type de projet il s'agit et comment cela fonctionne. Décrivez ce qui suit.

web: python /app/src/main.py

Il enseigne à heroku que le type de processus est Web et le programme qui l'exécute réellement.

J'ai également utilisé le heroku CLI pour le déploiement. Cette

$ git push heroku master

Vous pourrez vous déployer sur heroku.

la fin

Si vous avez des questions, des questions, des demandes d'amélioration ou des erreurs, veuillez nous en informer via Twitter.

De plus, GitHub a tout le code source. Code source