В третей части серии статей по написанию телеграм бота на python, мы настроим работу с внешним API. Бот будет запрашивать результаты матчей, преобразовывать в сообщение и выводить пользователю.
Выбор API для результатов матчей
Обычно я использую Rapid API для получения данных, там много бесплатных предложений. Под нашу задачу хорошо подходит Football Pro. Они дают 100 запросов в день, и возможность получить все результаты за раз.
Зарегистрируйтесь на Rapid Api, создайте приложение и оформите подписку на базовый (бесплатный) план. Сервис бесплатный, но для продолжения требуется карта.
Отлично, теперь можно следить за Лигой Чемпионов.
Получение данных с внешнего API
В прошлой части руководства я заложил будущую логику, хранение результатов по трем лигам в одном ключе. Так как у нас всего 10 лиг и нет разницы в запросе по трем лигам или всем сразу лучше результаты по каждой хранить отдельно. Это сэкономит запросы.
Где взять Api id и Api hash
Посмотрим в каком виде приходят данные в ответе с помощью интерфейса сервиса. Во вкладке «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 «) 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 limit_control(resp.headers) if resp.status_code == 200: return resp.json() else: logging.warning(f»Data retrieval error []. Headers: «) return #.
Для контроля бесплатных запросов я добавил функцию 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».
Я дописал в функцию 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-.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» n» for m in lg_matches: text += f»7> » if m[‘localteam_id’] == m[‘winner_team_id’]: text += f»** » else: text += f» » if m[‘time’][‘minute’] is not None: text += f»- » else: text += «— » if m[‘scores’][‘ht_score’] is not None: text += f»() » if m[‘visitorteam_id’] == m[‘winner_team_id’]: text += f»**n» else: text += f»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_»)), parse_mode=types.ParseMode.MARKDOWN, reply_markup=s.results_kb(user_leagues) )
Запустим и проверим, как работает бот:
TODO для вас.
1. Получение данных может длится несколько секунд, добавьте chat_action . Это текст, который отображается сверху во время выполнения кода.
2. Не всегда обновление результатов меняет сообщение, это приводит к ошибке. Пусть сообщение не редактируется, если текст дублирует старый.
Теперь допишем немного тестов и пойдем деплоить.
Тестирование бота
Будем проверять работоспособность API и контроль лимита.
Источник: pythonru.com
Создание и развертывание ретранслятора Telegram каналов, используя Python и Heroku
Для случая зеркалирования «один к одному» у нас все готово и можно переходить к развертыванию.
Зеркалирование «много к одному»
Когда каналов-источников и каналов, в которые необходимо пересылать сообщения, больше чем один, необходимо задать их взаимное соответствие.
Задавать карту соответствий прямо в коде — не вариант, поэтому для значения новой переменной окружения (CHANNELS_MAPPING) введем специальный формат записи:
Развертывание
Для развертывания приложения Heroku необходимо сделать следующее:
1. Зарегистрироваться на Heroku, если это не было сделано ранее;
3. Создать Heroku Procfile:
8 комментариев
Написать комментарий.
Полезная штука. Тут чувак сделал наподобие, только управление редиректами сделано через бота https://github.com/rumble-key/feed-bot-telegram
Развернуть ветку
классный гайд, можешь написать в личку — есть подобная задача, вдруг возможно сотрудничество с тобой!
Развернуть ветку
Спасибо! Настроил себе все)
Развернуть ветку
Подскажите, а почему позникает ошибка «Cant adapt type message»?
Развернуть ветку
Отличная работа! А вы не думали добавить фильтрацию измененных/удаленных сообщений?
Пример: добавляем новые сообщения в базу, и мониторим их. Если такое сообщение будет изменено/удалено — только тогда отправляем в «зеркальный» канал
Развернуть ветку
По ссылке в конце статьи находится обновленный проект, где можно добиться такого поведения изменив реплицирующие методы в EventProcessor’e ( https://github.com/khoben/telemirror/blob/a555136d3844381016916794ff8c78b9879eebf6/telemirror/mirroring.py#L16 ): на событие NewMessage только пишем в БД без отправки, на остальные события по наличию в БД отправляем сообщение.
Развернуть ветку
Комментарий удален модератором
Развернуть ветку
Heroku из России уже не доступен (не оплатить и т.д.). Можно все то-же самое сделать на Amvera Cloud, это отечественный аналог.
Развернуть ветку
Heroku был хорош тем, что там можно было бесплатно захостить небольшое приложение и поднять БД, но хорошие времена прошли. Если есть желание и время самому покопаться, то рекомендую поднять Dokku в виртуальном облаке, например, на самом дешевом тарифе от timeweb за ~180₽. Или поискать зарубежные PaaS, где есть возможность запустить без оплаты.
Источник: vc.ru
Авторизация пользователей через Telegram
Недавно Telegram добавил поддержку виджета для авторизации пользователей на сайте. Мы решили поэкспериментировать с ним и составить простую инструкцию, как настроить такую авторизацию самостоятельно.
В качестве примера будем использовать код на PHP, однако, данные шаги актуальны и для других языков программирования.
Настройка бота
Для использования виджета вам понадобится Telegram-бот.
Скопируйте токен бота, через которого вы хотите производить авторизацию пользователей.
Название и аватарка выбранного вами бота будут показаны пользователю во всплывающем окне. А вы получите возможность отправлять пользователю личные сообщения через этого бота.
Настройка виджета
На сайте можно получить код виджета и выбрать его внешний вид. К сожалению, возможностей для его произвольного конфигурирования на данный момент нет т.к. виджет встраивается на сайт посредством iframe.
Встраивание на сайт
После того, как пользователь нажмёт на кнопку, Telegram готов отправить вам данные любым из двух способов:
- Отправить пользователя на ваш сайт путём редиректа, передав информацию о нём в GET параметрах.
- Вызвать JavaScript функцию, передав в неё информацию о пользователе в качестве аргументов.
На данный момент поддерживаются следующие данные о пользователе:
- id – уникальный идентификатор пользователя в Telegram
- first_name, last_name – фамилия и имя из профиля пользователя
- username – уникальное имя из профиля
- photo_url – ссылка на аватарку пользователя в виде https://t.me/i/. /user.jpg
- auth_date – дата авторазации
- hash – HMAC-подпись ответа на основе секретного токена бота
Получение данных через JavaScript callback
Выберите в конструкторе виджета опцию Authorization Type: Callback. Сгенерированный в результате код виджета содержит JavaScript функцию, которая будет вызвана после успешной авторизации.
Эту функцию нужно передать в аттрибуте data-onauth тега
Вы можете произвольным образом реализовать функцию onTelegramAuth. Например, послать AJAX запрос на сервер с полученными аргументами.
Получение данных через Redirect
Выберите в конструкторе виджета опцию Authorization Type: Redirect to URL и введите URL, на который вы хотите получить запрос с данными пользователя. Например, введите адрес http://example.com/auth/telegram.
На странице обработки можно положить скрипт index.php следующего содержания:
Проверка данных пользователя
Чтобы удостовериться в правильности полученных данных, нужно проверить hash. Разработчики Telegram приводят пример кода проверки, добавим эту функцию в код из файла index.php
function checkTelegramAuthorization($auth_data) < $check_hash = $auth_data[‘hash’]; unset($auth_data[‘hash’]); $data_check_arr = []; foreach ($auth_data as $key =>$value) < $data_check_arr[] = $key . ‘=’ . $value; >sort($data_check_arr); $data_check_string = implode(«n», $data_check_arr); $secret_key = hash(‘sha256’, BOT_TOKEN, true); $hash = hash_hmac(‘sha256’, $data_check_string, $secret_key); if (strcmp($hash, $check_hash) !== 0) < throw new Exception(‘Data is NOT from Telegram’); >if ((time() — $auth_data[‘auth_date’]) > 86400) < throw new Exception(‘Data is outdated’); >return $auth_data; >
Разберём механизм работы функции проверки. В качестве аргумента она получает массив с данными пользователя.
array(7) < [«id»]=>string(7) «1831337» [«first_name»]=> string(18) «Александр» [«last_name»]=> string(16) «Менщиков» [«username»]=> string(5) «n0str» [«photo_url»]=> string(36) «https://t.me/i/userpic/100/n0str.jpg» [«auth_date»]=> string(10) «1518168109» [«hash»]=> string(64) «abba<..>1345» >
На первом шаге из массива извлекается значение по ключу hash и сохраняется в переменной.
На втором шаге массив преобразуется к виду key=value и сортируется в лексикографическом порядке. Полученные данные склеиваются в одну строку через разделитель “n” (код символа – 0xA0).
Далее происходит проверка равенства HMAC-SHA-256 подписи этой строки и значения сохранённого hash. Дополнительно проверяется не устарела ли auth_date.
В случае успеха, функция возвращает исходный массив без параметра hash.
Авторизация пользователя на сайте
Добавим в файл код вызова функции проверки
if (isset($_GET[‘hash’])) < try < $auth_data = checkTelegramAuthorization($_GET); echo «Hello, » . $auth_data[‘username’]; >catch (Exception $e) < die ($e->getMessage()); > >
Пользователь увидит сообщение с приветствием в случае успешной авторизации. Теперь вы можете сохранить информацию о нём в базу данных и привязать его ID к текущей сессии.
Пример кода из рабочего проекта
try < $profile = $tg->checkTelegramAuthorization($_GET); $id = $profile[‘id’]; $user = Model_User::findByAttribute(‘telegram_id’, $id); if ($user->is_empty()) < $user = new Model_User(); $user->telegram_id = $id; . $user->save() > else < . >>
Кастомизация кнопки
Сейчас из-за ограничений iframe нельзя изменить внешний вид кнопки. Однако, если возникла сильная необходимость, можно обойти это ограничения с помощью clickjacking.