[PYTHON] The story of making a Line Bot that tells us the schedule of competitive programming
Updated the Line Bot to notify you of competition programming dates. I also used DB etc. so I will summarize it a little.
What I made
If you send a sentence containing the word "contest" to the bot itself or the group that contains the bot, They will send you the Codeforces and AtCoder contest dates.
By the way, you can register as a friend from the QR code below.
Please use it.
What I used
-
Python Used as main language
-
PostgreSQL Used as a database
-
ngrok Used for testing bots in a local environment
-
heroku Used in production environment The add-on utilizes PostgreSQL and Scheduler.
-
Line Bot SDK Bot development kit provided by Line official
How it works
Get AtCoder Contest Dates
AtCoder doesn't have an official contest schedule API, so I'm scraping the official page to schedule a Rated contest. I'm using Python's BeautifulSoup for scraping. First, get the URL information and scrape it. After that, only the necessary data is taken out and formatted.
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
Get Codeforces dates
Codeforces has an Official API, so I hit the API to format it.
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 I'm using PostgreSQL, the official add-on for heroku. When inserting or retrieving a record, a function generates an SQL statement in advance and executes it. I am using a package called 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
I also update the contents of the DB once an hour on heroku's scheduler.
Sending data to Line
I used Flask to create a web application, as provided by the official SDK. The callback function is first called when the application starts. The handle_message function is called inside the callback function to send the message. Flex Message is used to send the message. Since Flex Message is generated in Json format, stock the Json template in advance and use it to send the message.
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 = 'Contest'
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')
Deploy to heroku
First, create a Procfile in your project. This tells heroku what type of project it is and how it works. Describe the following.
web: python /app/src/main.py
I'm teaching heroku that the process type is Web and the program that actually runs it.
I also used the heroku CLI for deployment. This
$ git push heroku master
Will allow you to deploy to heroku.
the end
If you have any questions, questions, improvement requests, or mistakes, please let us know via Twitter.
Also, all the source code is on GitHub. Source code
Recommended Posts