[PYTHON] Die Geschichte, einen Line Bot zu erstellen, der uns den Zeitplan für die Wettbewerbsprogrammierung erzählt

Der Line Bot wurde aktualisiert, um Sie über die Programmdaten des Wettbewerbs zu informieren. Ich habe auch DB usw. verwendet, daher werde ich es ein wenig zusammenfassen.

Was ich gemacht habe

Wenn Sie einen Satz mit dem Wort "Wettbewerb" an den Bot selbst oder an die Gruppe senden, die den Bot enthält, Sie senden Ihnen die Daten des Codeforces- und AtCoder-Wettbewerbs.

Übrigens können Sie sich über den unten stehenden QR-Code als Freund registrieren. Bitte benutzen Sie es. Screen Shot 2020-05-15 at 23.28.15.png

Was ich benutzt habe

Python Wird als Hauptsprache verwendet
PostgreSQL Wird als Datenbank verwendet
ngrok Wird zum Testen von Bot in lokaler Umgebung verwendet
heroku Wird für die Produktionsumgebung verwendet Das Add-On verwendet PostgreSQL und Scheduler.
Line Bot SDK Bot-Entwicklungskit von Line Official zur Verfügung gestellt

Wie es funktioniert

Holen Sie sich AtCoder-Wettbewerbsdaten

AtCoder hat keine offizielle API für den Wettbewerbsplan Ich kratzte auf der offiziellen Seite, um einen bewerteten Wettbewerb zu planen. Ich benutze Pythons BeautifulSoup zum Scraping. Holen Sie sich zuerst die URL-Informationen und kratzen Sie sie ab. Danach werden nur die erforderlichen Daten herausgenommen und formatiert.

`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

Erhalten Sie Codeforces-Daten

Codeforces hat eine offizielle API, daher habe ich die API aufgerufen, um sie zu formatieren.

`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 Ich verwende PostgreSQL, das offizielle Add-On von Heroku. Beim Einfügen oder Abrufen eines Datensatzes generiert eine Funktion im Voraus eine SQL-Anweisung und führt sie aus. Ich benutze ein Paket namens 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

Außerdem aktualisiert der Heroku-Scheduler den Inhalt der Datenbank einmal pro Stunde.

Daten an Line senden

Ich habe Flask verwendet, um eine Webanwendung zu erstellen, die vom offiziellen SDK bereitgestellt wird. Die Rückruffunktion wird zum ersten Mal aufgerufen, wenn die Anwendung gestartet wird. Die Funktion handle_message wird innerhalb der Rückruffunktion aufgerufen, um die Nachricht zu senden. Flex Message wird zum Senden von Nachrichten verwendet. Da Flex Message im Json-Format generiert wird, sollten Sie Json-Vorlagen im Voraus speichern und zum Senden von Nachrichten verwenden.

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

In Heroku bereitstellen

Erstellen Sie zunächst ein Procfile in Ihrem Projekt. Dies sagt Heroku, um welche Art von Projekt es sich handelt und wie es funktioniert. Beschreiben Sie Folgendes.

web: python /app/src/main.py

Es lehrt Heroku, dass der Prozesstyp Web ist und das Programm, das ihn tatsächlich ausführt.

Ich habe auch die heroku CLI für die Bereitstellung verwendet. Dies

$ git push heroku master

Sie können Heroku bereitstellen.

das Ende

Wenn Sie Fragen, Fragen, Verbesserungswünsche oder Fehler haben, teilen Sie uns dies bitte über Twitter mit.

Außerdem hat GitHub den gesamten Quellcode. Quellcode