Soccer Telegram bot in Python (2/4): Bot functionality

by Alex
Soccer Telegram bot in Python (2/4): Bot functionality

In the second part of our series on writing a python telegram bot, we will add functionality. The bot will welcome new users, offer to select and save leagues. We will add the ability to get results on selected leagues.

the “fish” of the bot code

Immediately write functions into “bot.py” that will be needed. Remove test_message beforehand:

# fonlinebot/app/bot.py
# ...
dp.middleware.setup(LoggingMiddleware())
@dp.message_handler(commands=['start'])
async def start_handler(message: types.Message):
"""Start command processing. Text and menu output"""
...
@dp.message_handler(commands=['help'])
async def help_handler(message: types.Message):
"""Help command processing. Text and menu output"""
...
@dp.callback_query_handler(lambda c: c.data == 'main_window')
async def show_main_window(callback_query: types.CallbackQuery):
"""main_window""
...
@dp.message_handler(lambda message: message.text == msg.btn_online)
@dp.message_handler(commands=['online'])
async def get_results(message: types.Message):
"""Processing commands online and the Online button.
Match request. Output results"""
...
@dp.callback_query_handler(lambda c: c.data.startswith('update_results'))
async def update_results():
"""update results message"""
...
@dp.message_handler(lambda message: message.text == msg.btn_config)
async def get_config(message: types.Message):
"""Processing Settings button.
Checking the selection of leagues. Display settings change menu"""
...
@dp.callback_query_handler(lambda c: c.data.startswith('edit_config'))
async def set_or_update_config(user_id: str):
"""Get or update selected leagues"""
...
@dp.callback_query_handler(lambda c: c.data[:6] in ['del_le', 'add_le'])
async def update_leagues_info(callback_query: types.CallbackQuery):
"""Add/delete league from cache, update message"""
...
@dp.callback_query_handler(lambda c: c.data == 'save_config')
async def save_config(callback_query: types.CallbackQuery):
"""Save user to database"""
....
@dp.callback_query_handler(lambda c: c.data == 'delete_config')
async def delete_config(user_id: str):
"""Deleting a user from the database""
...
@dp.message_handler()
async def unknown_message(message: types.Message):
"""Response to any unexpected message"""
...
async def on_shutdown(dp):
# ...

This is not the final version, I may have missed something. We’ll add the missing ones in the process. Each feature is wrapped with a decorator, that’s how we communicate with Telegarm:

  • @dp.message_handler(commands=['start']) – the decorator waits for message commands (which start with /). In this example, it’s waiting for the /start command.
  • @dp.callback_query_handler(lambda c: c.data == 'main_window') – expects callback and takes lambda function to filter it. The callback is sent by inline buttons. In the example we’re expecting a callback with the value 'main_window'.
  • @dp.message_handler(lambda message: message.text == msg.btn_config) – this decorator is similar to the previous one, but it expects a message from the user. In the example we will process a message with the text from msg.btn_config.

So. The user presses the start command, receives a welcome message. In it, we offer to select 3 leagues and monitor the results for them. You can get the results by command or by pressing the menu button. We also give the opportunity to change the selected competitions or delete your data from the bot. The full bot code from this tutorial is on gitlab.

Adding commands to the bot

Initially, commands are not configured. Users can enter them, but there is no special menu. To add them, you need to write https://t.me/botfather command /setcommands again. Select your bot and add this text:

start - Start and restart the bot
help - Bot options
online - Results of matches

In reply you will get “Success! Command list updated. /help”. Now you can go to your bot and check:  Добавление команд в бота

Command responses

Interaction with the bot begins with the /start command. You need to welcome and suggest the next step. This command will return a keyboard text. The same way /help works. Let’s add the handling of these commands to “bot.py”, update start_handler help_handler:

# fonlinebot/app/bot.py
# ...
from config import TOKEN, YEAR, MINUTE
import app.service as s
# ...
@dp.message_handler(commands=['start'])
async def start_handler(message: types.Message):
"""Start command processing. Display text and menu""""
# check if there is a user in the database
user_league_ids = await s.get_league_ids(message.from_user.id)
if not user_league_ids:
await message.answer(msg.start_new_user)
# adding the id of the message
cache.setex(f "last_msg_{message.from_user.id}", YEAR, message.message_id+2)
await set_or_update_config(user_id=message.from_user.id)
else:
await message.answer(msg.start_current_user,
reply_markup=s.MAIN_KB)
@dp.message_handler(commands=['help'])
async def help_handler(message: types.Message):
"""Help command processing. Text and menu output"""
await message.answer(msg.help, reply_markup=s.MAIN_KB)
# ...

I added import app.service as s. In this module the keyboard and the user validation function. the start_handler checks if the user is in the cache or the database, and sends him the appropriate text. Before sending the text to select leagues, I saved his future id. Got the number of the last message (that’s the “start” command itself) and added 2 points: +1 for our response to the command and +1 for the league selection message itself. Knowing the message id, it can be edited. Now let’s write the keyboard and get_league_ids to the module “service”.

# fonlinebot/app/service.py
from aiogram.types import ReplyKeyboardMarkup, KeyboardButton, \
InlineKeyboardMarkup, InlineKeyboardButton
from emoji import emojize
from config import BOT_LEAGUES, BOT_LEAGUE_FLAGS
from database import cache, database as db
from app.dialogs import msg
MAIN_KB = ReplyKeyboardMarkup(
resize_keyboard=True,
one_time_keyboard=True
).row(
KeyboardButton(msg.btn_online),
KeyboardButton(msg.btn_config)
)
async def get_league_ids(user_id: str) -> list:
"""The function gets the id of the user leagues in the database."""
leagues = cache.lrange(f "u{user_id}", 0, -1)
if leagues is None:
leagues = await db.select_users(user_id)
if leagues is not None:
leagues = leagues.split(',')
[cache.lpush(f "u{user_id}", lg_id) for lg_id in leagues]
else:
return []
return leagues

Ответы на команды MAIN_KB is the main keyboard, as in the screenshot above. Let’s break it down in more detail:

  • ReplyKeyboardMarkup is the object that creates the keyboard.
  • The parameter resize_keyboard=True reduces its size.
  • And with one_time_keyboard=True the keyboard will be hidden after use.
  • .row is a method to group the buttons in a row.
  • KeyboardButton(msg.btn_online) and KeyboardButton(msg.btn_config) are buttons with the specified text.

All that’s left to do is add the message texts to the dialogs. Insert this code into the Messages class.

# fonlinebot/app/dialogs.py
# ...
start_new_user: str = "Hi. I can let you know the results of the matches online."
start_current_user: str = "Hi. Welcome back! " \
"Use the commands or menu below to continue."
help: str = """
This bot gets match results for the last 48 hours.
Including LIVE mode.
- To select/change leagues press "Settings".
- To check the results, click "Online".
The bot was created for training purposes, for pythonru.com
"""

Selecting, changing and deleting leagues

Выбор, изменение и удаление лиг First let’s make edits to our auxiliary modules. In “dababase” we will add a new method insert_or_update_users to the Database class.

# fonlinebot/database.py
#...
async def insert_or_update_users(self, user_id: int, leagues: str):
user_leagues = await self.select_users(user_id)
if user_leagues is not None:
await self.update_users(user_id, leagues)
else:
await self.insert_users(user_id, leagues)
#...

Let’s add time metric variables to the settings:

# fonlinebot/config.py
#...
MINUTE = 60
YEAR = 60*60*24*366

And let’s add some text for the settings block:

# fonlinebot/app/dialogs.py
# ...
league_row: str = "{i}. {flag} {name}"
config: str = "currently selected:\n{leagues}"
btn_back: str = ""
btn_save: str = "Save"
config_btn_edit: str = "Change"
config_btn_delete: str = "delete data"
data_delete: str = "Data successfully deleted"
set_leagues: str = "Choose 3 leagues to track.\nSelected:\n{leagues}"
main: str = "What do we do?"
db_saved: str = "Settings saved."
cb_not_saved: str = "No Leagues Selected"
cb_limit: str = "Exceeded limit. Maximum 3 leagues."
# ...

This is the end of the preparation, it’s time to write the logic for adding leagues. In the module “service” let’s add two inline keyboards. Inline-keys are tied to a specific message. They can be used to send and process signals to the bot.

# fonlinebot/app/service.py
# ...
CONFIG_KB = InlineKeyboardMarkup().row(
InlineKeyboardButton(msg.btn_back, callback_data='main_window'),
InlineKeyboardButton(msg.config_btn_edit, callback_data='edit_config#')
).add(InlineKeyboardButton(msg.config_btn_delete, callback_data='delete_config'))
def leagues_kb(active_leagues: list, offset: int = 0):
kb = InlineKeyboardMarkup()
league_keys = list(BOT_LEAGUES.keys())[0+offset:5+offset]
for lg_id in league_keys:
if lg_id in active_leagues:
kb.add(InlineKeyboardButton(
f"{emojize(':white_heavy_check_mark:')} {BOT_LEAGUES[lg_id]}",
callback_data=f'del_league_#{offset}#{lg_id}'
))
else:
kb.add(InlineKeyboardButton(
BOT_LEAGUES[lg_id],
callback_data=f'add_league_#{offset}#{lg_id}'
))
kb.row(
InlineKeyboardButton(
msg.btn_back if offset else msg.btn_go,
callback_data="edit_config#0" if offset else "edit_config#5"),
InlineKeyboardButton(msg.btn_save, callback_data="save_config")
)
return kb
# ...

There are 3 buttons in the CONFIG_KB keyboard. The InlineKeyboardButton class takes the text and the callback_data parameter. It is the callback that Telegram will send us after you click on the button. This is what this keyboard looks like: Inline-клавиатура And leagues_kb generates a more complex keyboard, with pagination. We output 5 leagues, a “next/back” button, and a “save” button. The function takes the selected leagues and the indent. The indent is needed to output the leagues page by page. When user clicks on the league, we add it to the cache and when user clicks again we remove it. Please note, I’m generating dynamic line in callback_data. I insert parameters offset and lg_id, what to use in processing. Now let’s write function to nicely output list of leagues and update this list in cache:

# fonlinebot/app/service.py
# ...
async def get_league_names(ids: list) -> str:
"""The function collects a message with league names from the id"""
leagues_text = ""
for i, lg_id in enumerate(ids, start=1):
if i != 1:
leagues_text += '\n'
leagues_text += msg.league_row.format(
i=i,
flag=emojize(BOT_LEAGUE_FLAGS.get(lg_id, '-')),
name=BOT_LEAGUES.get(lg_id, '-')
)
return leagues_text
def update_leagues(user_id: str, data: str):
"""The function adds or removes the league id for the user"""
league_id = data.split("#")[-1] # data ~ add_league_#5#345
if data.startswith("add"):
cache.lpush(f "u{user_id}", league_id)
else:
cache.lrem(f "u{user_id}", 0, league_id)

Setting up the communication with Telegram

In the bot file let’s add functions for the settings menu.

# fonlinebot/app/bot.py
@dp.callback_query_handler(lambda c: c.data == 'main_window')
async def show_main_window(callback_query: types.CallbackQuery):
"""main_window""
await callback_query.answer()
await bot.send_message(callback_query.from_user.id, msg.main, reply_markup=s.MAIN_KB)
dp.message_handler(lambda message: message.text == msg.btn_config)
async def get_config(message: types.Message):
"""Processing Settings button.
Checking the selection of leagues. Display settings change menu""""
user_league_ids = await s.get_league_ids(message.from_user.id)
if user_league_ids:
cache.setex(f "last_msg_{message.from_user.id}", YEAR, message.message_id+2)
leagues = await s.get_league_names(user_league_ids)
await message.answer(msg.config.format(leagues=leagues),
reply_markup=s.CONFIG_KB)
else:
cache.setex(f "last_msg_{message.from_user.id}", YEAR, message.message_id+1)
await set_or_update_config(user_id=message.from_user.id)
@dp.callback_query_handler(lambda c: c.data == 'delete_config')
async def delete_config(callback_query: types.CallbackQuery):
"""Deleting a user from the database""
await db.delete_users(callback_query.from_user.id)
cache.delete(f "u{callback_query.from_user.id}")
await callback_query.answer()
cache.incr(f "last_msg_{callback_query.from_user.id}")
await bot.send_message(callback_query.from_user.id,
msg.data_delete,
reply_markup=s.MAIN_KB)

The get_config function is called after clicking on “Settings”. If the user has the leagues selected, it returns the standard message and settings menu. Otherwise it will call set_or_update_config to select leagues. At the same time I added data deletion and the home screen. We will create and edit the list in the same function. It will need parameters from the callback line. For example, the user pressed “forward ->” and Telegram sent "edit_config#5". We divided the string by # and took the last value (‘5’). This is how parameters will be passed between messages.

# fonlinebot/app/bot.py
@dp.callback_query_handler(lambda c: c.data.startswith('edit_config'))
async def set_or_update_config(callback_query: types.CallbackQuery = None,
user_id=None, offset=""):
"""Get or update selected leagues"""
# if callback came, get the data
if callback_query is not None:
user_id = callback_query.from_user.id
offset = callback_query.data.split("#")[-1]
league_ids = await s.get_league_ids(user_id)
leagues = await s.get_league_names(league_ids)
# if this is the first call of the function, we send a message
# if not, edit message and keyboard
if offset == "":
await bot.send_message(
user_id,
msg.set_leagues.format(leagues=leagues),
reply_markup=s.leagues_kb(league_ids)
)
else:
msg_id = cache.get(f "last_msg_{user_id}")
await bot.edit_message_text(
msg.set_leagues.format(leagues=leagues),
user_id,
message_id=msg_id
)
await bot.edit_message_reply_markup(
user_id,
message_id=msg_id,
reply_markup=s.leagues_kb(league_ids, int(offset))
)

What’s still missing is the response to clicking on the name of the leagues and the “save” button.

# fonlinebot/app/bot.py
dp.callback_query_handler(lambda c: c.data[:6] in ['del_le', 'add_le'])
async def update_leagues_info(callback_query: types.CallbackQuery):
"""Add/delete league from cache, update message"""
offset = callback_query.data.split("#")[-2]
s.update_leagues(callback_query.from_user.id, callback_query.data)
await set_or_update_config(user_id=callback_query.from_user.id, offset=offset)
await callback_query.answer()
@dp.callback_query_handler(lambda c: c.data == 'save_config')
async def save_config(callback_query: types.CallbackQuery):
"""Save user to database"""
leagues_list = await s.get_league_ids(callback_query.from_user.id)
if len(leagues_list) > 3:
# do not save if league limit is exceeded
await callback_query.answer(msg.cb_limit, show_alert=True)
elif leagues_list:
await db.insert_or_update_users(
callback_query.from_user.id,
",".join(leagues_list)
)
await callback_query.answer()
await bot.send_message(
callback_query.from_user.id,
msg.db_saved,
reply_markup=s.MAIN_KB
)
else:
# do not save if the list is empty
await callback_query.answer(msg.cb_not_saved)

Done. Now you can run the bot and add leagues in the settings.

Message with match results

Since the API is not yet configured, we’ll write a stub for these processes.

# fonlinebot/app/bot.py
@dp.message_handler(lambda message: message.text == msg.btn_online)
@dp.message_handler(commands=['online'])
async def get_results(message: types.Message):
"""Processing commands online and the Online button.
Match request. Output results"""
user_leagues = await s.get_league_ids(message.from_user.id)
cache.setex(f "last_msg_{message.from_user.id}", YEAR, message.message_id+1)
if not user_leagues:
await set_or_update_config(user_id=message.from_user.id)
else:
answer = await s.generate_results_answer(user_leagues)
cache.setex(f "last_update_{message.from_user.id}", MINUTE, "Updated")
await message.answer(answer, reply_markup=s.results_kb(user_leagues))
@dp.callback_query_handler(lambda c: c.data.startswith('update_results'))
async def update_results(callback_query: types.CallbackQuery):
"""update results message"""
if cache.get(f "last_update_{callback_query.from_user.id}") is None:
user_leagues = callback_query.data.split("#")[1:]
answer = await s.generate_results_answer(user_leagues)
cache.setex(f "last_update_{callback_query.from_user.id}", MINUTE, "Updated")
await bot.edit_message_text(
answer,
callback_query.from_user.id,
message_id=int(cache.get(f "last_msg_{callback_query.from_user.id}")
)
# ignore the update if less than a minute has passed
await callback_query.answer(msg.cb_updated)

get_results gets the leagues and fixes the message id for editing. If user doesn’t have saved leagues – it calls set_or_update_config, in other case we generate response with matches and “update” button. Футбольный телеграм бот на Python (2/4): Функциональность бота The update results function only performs logic if more than a minute has passed since the last update. For this purpose, after each result response we add a record to the cache with a retention time of 1 minute. Let’s add the dependencies:

# fonlinebot/app/dialogs.py
#...
results: str = "all results for today\n{matches}"
no_results: str = "no matches today"
update_results: str = "Update results"
cb_updated: str = f"{emojize(':white_heavy_check_mark:')} Done"
#...

Update button and response generation in “service.py”.

# fonlinebot/app/service.py
from config import BOT_LEAGUES, BOT_LEAGUE_FLAGS, MINUTE
#...
def results_kb(leagues: list):
params = [f "#{lg}" for lg in leagues]
kb = InlineKeyboardMarkup()
kb.add(InlineKeyboardButton(
msg.update_results,
callback_data=f "update_results{''.join(params)}"
))
return kb
async def generate_results_answer(ids: list) -> str:
"""The function generates a message to output match results"""
results = await get_last_results(ids)
if results:
text_results = results_to_text(results)
return msg.results.format(matches=text_results)
else:
return msg.no_results
def ids_to_key(ids: list) -> str:
"""Standardization of keys for storing matches"""
ids.sort()
return ",".join(ids)
async def parse_matches(ids: list) -> list:
"""API match retrieval function"""
# we'll write the logic in the next part
return []
async def get_last_results(league_ids: list) -> list:
lg_key = ids_to_key(league_ids)
last_results = cache.jget(lg_key)
if last_results is None:
last_results = await parse_matches(league_ids)
if last_results:
# add new matches if there are any
cache.jset(lg_key, last_results, MINUTE)
return last_results
def results_to_text(matches: list) -> str:
"""
Function to generate a message with matches
"""
# we will write the logic in the next part
...
#...

The generate_results_answer function gets matches, converts the data to text and returns it. If there are no matches, it returns the corresponding text. To save resources we check availability of matches in the cache and only then apply to API.

Processing unknown messages

People will write text that we don’t process. Update the Messages class:

# fonlinebot/app/dialogs.py
#...
unknown_text: str = "I don't understand anything, but it's very interesting. Try /help."
#...

And unknown_message.

# fonlinebot/app/bot.py
#...
@dp.message_handler()
async def unknown_message(message: types.Message):
"""Answer to any unexpected message"""
await message.answer(msg.unknown_text, reply_markup=s.MAIN_KB)
#...

Let’s make sure that everything works: Футбольный телеграм бот на Python (2/4): Функциональность бота Great. We’ve written a bot with keyboards, caching, and a database. Now it’s time to add tests.

Updating the tests

Let’s add a TestService class to control the get_league_ids and get_last_results functions.

# fonlinebot/test.py
from app import bot, service
#...
class TestService(IsolatedAsyncioTestCase):
async def test_get_league_ids(self):
ids = await service.get_league_ids("1111")
self.assertEqual(type(ids), list)
async def test_get_last_results(self):
results = await service.get_last_results(["1", "2", "3"])
self.assertEqual(type(results), list)
#...

The link to the repository with the code is at the beginning of the article. Good luck!

What’s next?

There’s a small part of work left to do. In the next part, we’ll find the right API and set up the processing of getting the results: adding an external API.

Related Posts

LEAVE A COMMENT