Футбольный телеграм бот на Python (3/4): Получение внешних данных

В третей части серии статей по написанию телеграм бота на 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 лиг. Вместе с этим обновились и некоторые лиги. Запустим и проверим:

Футбольный телеграм бот на Python (3/4): Получение внешних данных

Отлично, теперь можно следить за Лигой Чемпионов.

Получение данных с внешнего 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)
Футбольный телеграм бот на Python (3/4): Получение внешних данных

На самом деле лимит начисляется каждые 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))

Бот готов! Код этого урока в начале статьи.

Дальше мы подготовим и запустим бота на удаленном сервере.

Максим
Я создал этот блог в 2018 году, чтобы распространять полезные учебные материалы, документации и уроки на русском. На сайте опубликовано множество статей по основам python и библиотекам, уроков для начинающих и примеров написания программ.
Мои контакты: Почта
admin@pythonru.comAlex Zabrodin2018-10-26OnlinePython, Programming, HTML, CSS, JavaScript