Soccer Telegram bot in Python (3/4): Getting external data

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

In part three of our series on writing a python telegram bot, we’re going to configure how to work with the external API. The bot will query match results, convert them into a message, and output them to the user. The full bot code from this tutorial is on gitlab.

Choosing the API for match results

I usually use Rapid API to retrieve data and there are a lot of free offerings. Football Pro is a good fit for our task. They give you 100 queries per day, and the ability to get all the results at once. Sign up for Rapid Api, create an app, and subscribe to the basic (free) plan. The service is free, but a card is required to continue. After you subscribe, you’ll receive a unique key, which we’ll add to your settings later. You can make only 100 requests per day for free. If you exceed the limit you will be charged. Although we will stop working when the limit is reached, I am not responsible for possible charges.

Updating API settings

Let’s add variables for requests. The key can be found on the “Endpoints” tab at the bottom in the “X-RapidAPI-Key” field. The rest of the lines can be copied from me. We will query the data at the specified address with headers for authorization and parameters for filtering.

# fonlinebot/config.py
# ...
SOCCER_API_URL = "https://football-pro.p.rapidapi.com/api/v2.0/livescores"
SOCCER_API_HEADERS = {
'x-rapidapi-key': 'your unique key',
{ 'x-rapidapi-host': 'football-pro.p.rapidapi.com'
}
SOCCER_API_PARAMS = {
"tz": "Europe/Moscow",
"include": "localTeam,visitorTeam"
}
# ...

In the same file we need to edit the data of the leagues. I suggest ready-made, you can choose others (Id here):

# fonlinebot/config.py
# ...
BOT_LEAGUES = {
"82": "German Bundesliga",
"384": "Italian Serie A",
"564": "Spanish La Liga",
"462": "Portuguese Primera Liga",
"72": "Dutch Championship",
"2": "Champions League",
"5": "Europa League",
"8": "English Premier League",
"301": "French Ligue 1",
"486": "Russian Premier League."
}
# Flags for posts, emoji-code
BOT_LEAGUE_FLAGS = {
"82": ":Germany:",
"384": ":Italy:",
"564": ":Spain:",
"462": ":Portugal:"
"72": ":Netherlands:",
"2": ":European_Union:",
"5": ":trophy:",
"8": ":England:",
"301": ":France:"
"486": ":Russia:"
}
# ...

Instead of test 1,2,3 I added real league id’s. Some leagues have been updated along with this. Let’s run it and check it out: Футбольный телеграм бот на Python (3/4): Получение внешних данных Great, now you can follow the Champions League.

Getting data from external API

In the last part of the tutorial I laid out the future logic, storing the results of the three leagues in one key. Since we only have 10 leagues and it doesn’t make a difference to query three leagues or all at once it’s better to store the results for each one separately. This will save queries. Let’s see how the data in the response using the service interface. In the tab “Endpoints” on the left, select “Fixtures of Today” and click “Test Endpoint”. The answer appears in the right column. данные в ответе These are the lines we will use for each match:

{
...
"league_id":998
...
"scores":{
...
"ht_score": "0-0"
"ft_score": "1-1"
...
}
}, "time":{
"status": "FT"
"starting_at":{
"time": "08:00:00"
...
}
"minute":90
...
"added_time":NULL
...
}
...
"localTeam":{
"data":{
...
"name": "Hadiya Hosaena"
...
}
}
}, "visitorTeam":{
"data":{
...
{ "name": "Kedus Giorgis"
...
}
}
}

To send requests, we need to install the requests library: pip install requests==2.25.1. Let’s write a function that makes a request to the API. Sometimes we will get errors in response, we must be prepared. Let’s send logs about the error and return it. TODO for you. Set up to send a message to admin if fetch_results returned a dictionary with an "error" key.

# fonlinebot/app/service.py
import requests
import logging
from config import BOT_LEAGUES, BOT_LEAGUE_FLAGS, MINUTE, 
SOCCER_API_URL, SOCCER_API_HEADERS, SOCCER_API_PARAMS
# ...
def limit_control(headers):
"""Free request limit control"""
if headers.get("x-ratelimit-requests-remaining") is None:
logging.error(f "Invalid headers response {headers}")
if int(headers['x-ratelimit-requests-remaining']) dict:
SOCCER_API_PARAMS['leagues'] = ",".join(BOT_LEAGUES.keys())
try:
resp = requests.get(SOCCER_API_URL,
headers=SOCCER_API_HEADERS,
params=SOCCER_API_PARAMS)
except requests.ConnectionError:
logging.error("ConnectionError")
return {"error": "ConnectionError"}
limit_control(resp.headers)
if resp.status_code == 200:
return resp.json()
else:
logging.warning(f "Data retrieval error [{resp.status_code}]. headers: {resp.headers} ")
return {"error": resp.status_code}
#...

I added limit_control function to control free queries. When there are less than 6 requests left, the corresponding entry will be added to the cache. Now the bot will check if this entry is in the cache before sending request. We will place the check in generate_results_answer. If there is an entry, we will return a warning.

# fonlinebot/app/service.py
#...
async def generate_results_answer(ids: list) -> str:
"""The function generates a message to output the results of the matches."""
limit = cache.get("limit_control")
if limit is not None:
return limit
results = await get_last_results(ids)
if results == [[]]*len(ids):
return msg.no_results
elif msg.fetch_error in results:
return msg.fetch_error
else:
text_results = results_to_text(results)
return msg.results.format(matches=text_results)
#...

Now update “bot.py” and “dialogs.py”.

# fonlinebot/app/bot.py
#...
@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)
if answer == msg.limit_control:
return await callback_query.answer(answer, show_alert=True)
else:
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}")),
reply_markup=s.results_kb(user_leagues)
)
# ignore the update if less than a minute has passed
await callback_query.answer(msg.cb_updated)
#...

I have added 2 lines to the function to check the answer. If the text in the answer exceeds the limit we show a warning.

# fonlinebot/app/dialogs.py
#...
limit_control: str = "Request limit has been reached. Come back tomorrow."
fetch_error: str = "Fetch error, try again later."
#...

You can add such an entry for a minute and make sure.

cache.setex("limit_control", 60, msg.limit_control)

Футбольный телеграм бот на Python (3/4): Получение внешних данных In fact, the limit is calculated every 24 hours from the time you signed up. If you subscribed at 1:00 p.m., that’s the time the balance is updated. You can get the remaining time in seconds in the response headers by the x-ratelimit-requests-reset key.

Clearing API data and saving

Now let’s write a function which unpacks the response to save it in the cache. TODO for you. It’s not always necessary to update the matches. For example we got the list at 8 am and the first match will start at 7 pm. Before the first match starts the results will not change, you can save queries here.

# fonlinebot/app/service.py
#...
async def parse_matches() -> dict:
"""API match collection function"""
data = {}
matches = fetch_results()
if matches.get("error", False):
return matches
for m in matches['data']:
if not data.get(str(m['league_id']), False):
data[str(m['league_id'])] = [m]
else:
data[str(m['league_id'])].append(m)
return data
#...

We call this function in get_last_results if we didn’t find any results in the cache. Let’s add the saving of last results there:

# fonlinebot/app/service.py
#...
async def save_results(matches: dict):
"""Saving match results""
for lg_id in BOT_LEAGUES.keys():
cache.jset(lg_id, matches.get(lg_id, []), MINUTE)
async def get_last_results(league_ids: list) -> list:
last_results = [cache.jget(lg_id) for lg_id in league_ids]
if None in last_results:
all_results = await parse_matches()
if all_results.get("error", False):
return [msg.fetch_error]
else:
await save_results(all_results)
last_results = [all_results.get(lg_id, []) for lg_id in league_ids]
return last_results
#...

If for some league we have no record, we call the API and save the result for 1 minute.

Logs saving to a file

There are important logs, which will help us to correct errors of receiving data. Let’s add logging settings: format and writing to a file.

# fonlinebot/config.py
#...
formatter = '[%(asctime)s] %(levelname)8s --- %(message)s (%(filename)s:%(lineno)s)'
logging.basicConfig(
# TODO uncommented on the server
# filename=f'bot-from-{datetime.datetime.now().date()}.log',
# filemode='w',
format=formatter,
datefmt='%Y-%m-%d %H:%M:%S',
# TODO logging.WARNING
level=logging.DEBUG
)
#...

Now the line will look like this. The time and place of the log appears:

[2021-02-05 11:38:29] INFO --- Database connection established (database.py:38)

There are many options for configuring logging, we won’t dwell on that in detail. Lessons are dedicated to the telegram bot.

Nice message output to the user

Ok, we have got a list of dictionaries with lots of data. It needs to be turned into text format:

English Premier League
Finished Tottenham 0:1 (0:1) Chelsea

The formatting is implemented in results_to_text.

# fonlinebot/app/service.py
#...
def add_text_time(time: dict) -> str:
"""Selecting text depending on match status
All statuses are here:
https://sportmonks.com/docs/football/2.0/getting-started/a/response-codes/85#definitions
"""
scheduled = ["NS"]
ended = ["FT", "AET", "FT_PEN"]
live = ["LIVE", "HT", "ET", "PEN_LIVE"]
if time['status'] in scheduled and time['starting_at']['time'] is not None:
# truncate the seconds
return time['starting_at']['time'][:-3]
elif time['status'] in finished:
# return "ended"
elif time['status'] in live and time['minute'] is not None:
if time['extra_minute'] is not None:
return time['minute'] + time['extra_minute']
return time['minute']]
else:
# for other statuses return stub
return "--:--"
def results_to_text(matches: list) -> str:
"""
Function to generate a message with matches
Gets list[list[dict]]
Returns the text:
| English Premier League |
| Ended Tottenham 0:1 (0:1) Chelsea |
...
"""
text = ""
for lg_matches in matches:
if not lg_matches:
continue
lg_flag = BOT_LEAGUE_FLAGS[str(lg_matches[0]['league_id'])
lg_name = BOT_LEAGUES[str(lg_matches[0]['league_id'])]
text += f"{emojize(lg_flag)} {lg_name}n"
for m in lg_matches:
text += f "{add_text_time(m['time']):>7} "
if m['localteam_id'] == m['winner_team_id']:
text += f "*{m['localTeam']['data']['name']}* "
else:
text += f"{m['localTeam']['data']['name']} "
if m['time']['minute'] is not None:
text += f"{m['scores']['localteam_score']}-{m['scores']['visiteam_score']} "
else:
text += "- "
if m['scores']['ht_score'] is not None:
text += f"({m['scores']['ht_score']}) "
if m['visitorteam_id'] == m['winner_team_id']:
text += f "*{m['visitorTeam']['data']['name']}*n"
else:
text += f"{m['visitorTeam']['data']['name']}n"
text += "n"
return text
#...

The function loops through the leagues and matches and generates a readable output. Before running it is necessary to add the parse_mode parameter to the messages, to bold the winning teams.

# fonlinebot/app/bot.py
#...
async def get_results(message: types.Message):
#...
await message.answer(answer,
reply_markup=s.results_kb(user_leagues),
parse_mode=types.ParseMode.MARKDOWN)
# ...
async def update_results(callback_query: types.CallbackQuery):
# ...
await bot.edit_message_text(
answer,
callback_query.from_user.id,
message_id=int(cache.get(f "last_msg_{callback_query.from_user.id}")),
parse_mode=types.ParseMode.MARKDOWN,
reply_markup=s.results_kb(user_leagues)
)

Let’s run and check how the bot works: TODO for you. 1. It may take a few seconds to get the data, add chat_action. This is the text that is displayed at the top during code execution. 2. Not always updating the results will change the message, it will cause an error. Let the message not be edited if the text duplicates the old one. Now let’s finish some tests and go to deplot.

Testing the bot

We will check the API and limit control.

# fonlinebot/test.py
#...
import requests
import config
#...
class TestService(IsolatedAsyncioTestCase):
#...
def test_limit_control(self):
test_data = {'x-ratelimit-requests-reset': '60',
'x-ratelimit-requests-remaining': "0"}
service.limit_control(test_data)
self.assertIsNotNone(cache.get("limit_control"))
class TestAPI(unittest.TestCase):
def test_api_response(self):
result = service.fetch_results()
self.assertIsNotNone(result.get('data', None))
def test_api_headers(self):
config.SOCCER_API_PARAMS['leagues'] = ",".join(config.BOT_LEAGUES.keys())
resp = requests.get(
config.SOCCER_API_URL,
headers=config.SOCCER_API_HEADERS,
params=config.SOCCER_API_PARAMS
)
self.assertIsNotNone(resp.headers.get('x-ratelimit-requests-reset', None))

Bot ready! The code of this lesson is at the beginning of the article. Next, we’ ll prepare and run the bot on a remote server.

Related Posts

LEAVE A COMMENT