В третей части серии статей по написанию телеграм бота на python, мы настроим работу с внешним API. Бот будет запрашивать результаты матчей, преобразовывать в сообщение и выводить пользователю.
Полный код бота из этого урока на gitlab.
Выбор API для результатов матчей
Обычно я использую Rapid API для получения данных, там много бесплатных предложений. Под нашу задачу хорошо подходит Football Pro. Они дают 100 запросов в день, и возможность получить все результаты за раз.
Зарегистрируйтесь на Rapid Api, создайте приложение и оформите подписку на базовый (бесплатный) план. Сервис бесплатный, но для продолжения требуется карта.
После подписки вы получите уникальный ключ, который мы позже добавим в настройки.
Бесплатно можно делать только 100 запросов в сутки. При превышении лимита с вас будут брать деньги. Хотя мы будем останавливать работу при достижении лимита, я не несу ответственность за возможные списания.
Обновление настроек для API
Добавим переменные для запросов. Ключ можно найти на вкладке «Endpoints» внизу в поле «X-RapidAPI-Key». Остальные строки можно копировать у меня. Мы будем запрашивать данные по указанному адресу с заголовками для авторизации и параметрами для фильтрации.
# fonlinebot/config.py
# ...
SOCCER_API_URL = "https://football-pro.p.rapidapi.com/api/v2.0/livescores"
SOCCER_API_HEADERS = {
'x-rapidapi-key': "ваш уникальный ключ",
'x-rapidapi-host': "football-pro.p.rapidapi.com"
}
SOCCER_API_PARAMS = {
"tz": "Europe/Moscow",
"include": "localTeam,visitorTeam"
}
# ...
В этом же файле нужно отредактировать данные лиг. Я предлагаю уже готовые, вы можете выбрать другие (Id здесь):
# fonlinebot/config.py
# ...
BOT_LEAGUES = {
"82": "Немецкая Бундеслига",
"384": "Итальянская Серия А",
"564": "Испанская Ла Лига",
"462": "Португальская Примейра Лига",
"72": "Чемпионат Нидерландов",
"2": "Лига Чемпионов",
"5": "Лига Европы",
"8": "Английская Премьер-лига",
"301": "Французская Лига 1",
"486": "Российская Премьер-лига"
}
# Флаги для сообщений, emoji-код
BOT_LEAGUE_FLAGS = {
"82": ":Germany:",
"384": ":Italy:",
"564": ":Spain:",
"462": ":Portugal:",
"72": ":Netherlands:",
"2": ":European_Union:",
"5": ":trophy:",
"8": ":England:",
"301": ":France:",
"486": ":Russia:"
}
# ...
Вместо тестовых 1,2,3 я добавил реальные id лиг. Вместе с этим обновились и некоторые лиги. Запустим и проверим:
Отлично, теперь можно следить за Лигой Чемпионов.
Получение данных с внешнего API
В прошлой части руководства я заложил будущую логику, хранение результатов по трем лигам в одном ключе. Так как у нас всего 10 лиг и нет разницы в запросе по трем лигам или всем сразу лучше результаты по каждой хранить отдельно. Это сэкономит запросы.
Посмотрим в каком виде приходят данные в ответе с помощью интерфейса сервиса. Во вкладке «Endpoints» слева выберем «Fixtures of Today» и нажмем «Test Endpoint». Ответ появится в правом столбце.
Вот эти строки мы будем использовать для каждого матча:
{
...
"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"
...
}
}
}Для отправки запросов нужно установить библиотеку requests: pip install requests==2.25.1.
Напишем функцию, которая делает запрос к API. Иногда в ответ мы будем получать ошибки, нужно быть готовым. Отправим логи об ошибке и вернем ее.
TODO для вас. Настройте отправку сообщения админу, если fetch_results вернула словарь с ключом "error".
# 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):
"""Контроль бесплатного лимита запросов"""
if headers.get("x-ratelimit-requests-remaining") is None:
logging.error(f"Invalid headers response {headers}")
if int(headers['x-ratelimit-requests-remaining']) <= 5:
cache.setex(
"limit_control",
int(headers['x-ratelimit-requests-reset']),
msg.limit_control
)
def fetch_results() -> 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}
#...
Для контроля бесплатных запросов я добавил функцию limit_control. Когда останется меньше 6ти запросов, в кеш добавится соответствующая запись. Теперь бот будет проверять наличие этой записи в кеше, прежде чем отправлять запрос.
Проверку разместим в generate_results_answer. Если запись есть, мы вернем предупреждение.
# fonlinebot/app/service.py
#...
async def generate_results_answer(ids: list) -> str:
"""Функция создaет сообщение для вывода результатов матчей"""
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)
#...
А теперь обновите «bot.py» и «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):
"""Обновление сообщения результатов"""
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)
)
# игнорируем обновление, если прошло меньше минуты
await callback_query.answer(msg.cb_updated)
#...
Я дописал в функцию 2 строки для проверки answer. Если в ответе текст превышения лимита мы показываем предупреждение.
# fonlinebot/app/dialogs.py
#...
limit_control: str = "Лимит запросов исчерпан. Возвращайтесь завтра."
fetch_error: str = "Ошибка получения данных, попробуйте позже."
#...
Можете добавить такую запись на минуту и убедиться.
cache.setex("limit_control", 60, msg.limit_control)
На самом деле лимит начисляется каждые 24 часа с момента подписки. Если вы подписались в 13:00, значит это время обновления остатка. В заголовках ответа по ключу
x-ratelimit-requests-resetможно получить остаток времени в секундах.
Очистка данных API и сохранение
Теперь напишем функцию которая распарсит ответ для сохранения в кеш.
TODO для вас. Не всегда нужно обновлять матчи. Например, мы в 8 утра получили список и первый матч начнется в 19.00. До начала первого матча результаты не изменятся, здесь можно сэкономить запросы.
# fonlinebot/app/service.py
#...
async def parse_matches() -> dict:
"""Функция сбора матчей по API"""
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
#...
Эту функцию мы вызываем в get_last_results, если не нашли результатов в кеше. Давайте туда допишем сохранение последних результатов:
# fonlinebot/app/service.py
#...
async def save_results(matches: dict):
"""Сохранение результатов матчей"""
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
#...
Если по какой-то лиге у нас нет записи, мы обращаемся к API и сохраняем результат на 1 минуту.
Запись логов в файл
Появились важные логи, которые помогут исправлять ошибки получения данных. Добавим настройки логирования: формат и запись в файл.
# fonlinebot/config.py
#...
formatter = '[%(asctime)s] %(levelname)8s --- %(message)s (%(filename)s:%(lineno)s)'
logging.basicConfig(
# TODO раскомментировать на сервере
# 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
)
#...
Теперь строка будет выглядеть так. Появилось время и место лога:
[2021-02-05 11:38:29] INFO --- Database connection established (database.py:38)Для настройки логирования много вариантов, мы не будет подробно на этом останавливать. Уроки посвящены телеграм боту.
Красивый вывод сообщения пользователю
Хорошо, мы получили список словарей с множеством данных. Его нужно превратить в текст формата:
Английская Премьер-лига Окончен Тоттенхэм 0:1 (0:1) Челси
Форматирование реализуем в results_to_text.
# fonlinebot/app/service.py
#...
def add_text_time(time: dict) -> str:
"""Подбор текста в зависимости от статуса матча
Все статусы здесь:
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:
# обрезаем секунды
return time['starting_at']['time'][:-3]
elif time['status'] in ended:
return "Окончен"
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:
# для других статусов возвращаем заглушку
return "--:--"
def results_to_text(matches: list) -> str:
"""
Функция генерации сообщения с матчами
Получает list[list[dict]]]
Возвращает текст:
| Английская Премьер-лига |
| Окончен Тоттенхэм 0:1 (0:1) Челси |
...
"""
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']['visitorteam_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
#...
Функция циклом проходит по лигам и матчам, формирует читаемый вывод. Перед запуском нужно добавить параметр parse_mode в сообщения, для выделения жирным команд-победителей.
# 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)
)
Запустим и проверим, как работает бот:
TODO для вас.
1. Получение данных может длится несколько секунд, добавьте chat_action. Это текст, который отображается сверху во время выполнения кода.
2. Не всегда обновление результатов меняет сообщение, это приводит к ошибке. Пусть сообщение не редактируется, если текст дублирует старый.
Теперь допишем немного тестов и пойдем деплоить.
Тестирование бота
Будем проверять работоспособность API и контроль лимита.
# 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))
Бот готов! Код этого урока в начале статьи.
