Во второй части серии статей по написанию телеграм бота на python, мы добавим функциональность. Бот будет приветствовать новых юзеров, предлагать выбрать и сохранять лиги. Добавим возможность получить результаты по выбранным лигам.
«Рыба» кода бота
Сразу запишем функции в «bot.py», которые понадобятся. Предварительно удалите test_message:
# fonlinebot/app/bot.py
# ...
dp.middleware.setup(LoggingMiddleware())
@dp.message_handler(commands=['start'])
async def start_handler(message: types.Message):
"""Обработка команды start. Вывод текста и меню"""
...
@dp.message_handler(commands=['help'])
async def help_handler(message: types.Message):
"""Обработка команды help. Вывод текста и меню"""
...
@dp.callback_query_handler(lambda c: c.data == 'main_window')
async def show_main_window(callback_query: types.CallbackQuery):
"""Главный экран"""
...
@dp.message_handler(lambda message: message.text == msg.btn_online)
@dp.message_handler(commands=['online'])
async def get_results(message: types.Message):
"""Обработка команды online и кнопки Онлайн.
Запрос матчей. Вывод результатов"""
...
@dp.callback_query_handler(lambda c: c.data.startswith('update_results'))
async def update_results():
"""Обновление сообщения результатов"""
...
@dp.message_handler(lambda message: message.text == msg.btn_config)
async def get_config(message: types.Message):
"""Обработка кнопки Настройки.
Проверка выбора лиг. Вывод меню изменений настроек"""
...
@dp.callback_query_handler(lambda c: c.data.startswith('edit_config'))
async def set_or_update_config(user_id: str):
"""Получение или обновление выбранных лиг"""
...
@dp.callback_query_handler(lambda c: c.data[:6] in ['del_le', 'add_le'])
async def update_leagues_info(callback_query: types.CallbackQuery):
"""Добавление/удаление лиги из кеша, обновление сообщения"""
...
@dp.callback_query_handler(lambda c: c.data == 'save_config')
async def save_config(callback_query: types.CallbackQuery):
"""Сохранение пользователя в базу данных"""
....
@dp.callback_query_handler(lambda c: c.data == 'delete_config')
async def delete_config(user_id: str):
"""Удаление пользователя из базы данных"""
...
@dp.message_handler()
async def unknown_message(message: types.Message):
"""Ответ на любое неожидаемое сообщение"""
...
async def on_shutdown(dp):
# ...
Это не окончательная версия, я мог что-то упустить. В процессе добавим недостающие.
Каждая функция обернута декоратором, так мы общаемся с Телегармом:
@dp.message_handler(commands=['start'])— декоратор ожидает сообщения-команды (которые начинаются с/). В этом примере он ожидает команду/start.@dp.callback_query_handler(lambda c: c.data == 'main_window')— ожидает callback и принимает lambda-функцию для его фильтрации. Callback отправляется inline-кнопками. В примере мы ожидаем callback со значением'main_window'.@dp.message_handler(lambda message: message.text == msg.btn_config)— этот декоратор похож на предыдущий, но ожидает сообщение от пользователя. В примере мы будем обрабатывать сообщение с текстом изmsg.btn_config.
Итак. Пользователь нажимает команду старт, получает приветственное сообщение. В нем мы предлагаем выбрать 3 лиги и мониторить результаты по ним. Получить результаты можно командой или кнопкой меню. Так же мы даем возможность изменить выбранные соревнования или удалить свои данные из бота.
Полный код бота из этого урока на gitlab.
Добавление команд в бота
Изначально команды не настроены. Пользователи могут вводить их, но специально меню нет. Для добавления нужно снова написать https://t.me/botfather команду /setcommands. Выберите своего бота и добавьте этот текст:
start - Запуск и перезапуск бота
help - Возможности бота
online - Результаты матчей
В ответ получите «Success! Command list updated. /help». Теперь можно перейти в своего бота и проверить:
Ответы на команды
Взаимодействие с ботом начинается с команды /start. Нужно поприветствовать и предложить следующий шаг. Эта команда будет возвращать текст с клавиатурой. Точно так же работает и /help.
Добавим обработку этих команд в «bot.py», обновите 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. Вывод текста и меню"""
# проверка, есть ли пользователь в базе
user_league_ids = await s.get_league_ids(message.from_user.id)
if not user_league_ids:
await message.answer(msg.start_new_user)
# добавление id сообщения настроек
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. Вывод текста и меню"""
await message.answer(msg.help, reply_markup=s.MAIN_KB)
# ...
Я добавил импорт import app.service as s. В этом модуле клавиатура и функция проверки пользователя. start_handler проверяет есть ли пользователь в кеше или базе данных, и отправляет ему соответствующий текст.
Перед отправкой текста для выбора лиг, я сохранил его будущий id. Получил номер последнего сообщения (это сама команда «start») и добавил 2 пункта: +1 за наш ответ на команду и +1 за само сообщения выбора лиг. Зная id сообщения, его можно редактирвать.
Теперь напишем клавиатуру и get_league_ids в модуль «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:
"""Функция получает id лиг пользователя в базе данных"""
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 — основная клавиатура, как на скриншоте выше. Разберем подробнее:
ReplyKeyboardMarkup— объект, который создает клавиатуру.- Параметр
resize_keyboard=Trueуменьшает ее размер. - А с
one_time_keyboard=Trueклавиатура будет скрываться, после использования. .row— метод для группировки кнопок в строку.KeyboardButton(msg.btn_online)иKeyboardButton(msg.btn_config)— кнопки с заданным текстом.
Осталось только добавить текста сообщений в dialogs. Вставьте этот код в класс Messages.
# fonlinebot/app/dialogs.py
# ...
start_new_user: str = "Привет. Я могу сообщать тебе результаты матчей online."
start_current_user: str = "Привет. С возвращением! " \
"Используй команды или меню внизу для продолжения."
help: str = """
Этот бот получает результаты матчей за последние 48 часов.
Включая режим LIVE.
- Что бы выбрать/изменить лиги нажмите "Настройки".
- Для проверки результатов нажмите "Онлайн".
Бот создан в учебных целях, для сайта pythonru.com
"""
Выбор, изменение и удаление лиг
Сначала внесем правки в наши вспомогательные модули.
В «dababase» в класс Database добавим новый метод insert_or_update_users.
# 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)
#...
В настройки добавим переменные метрик времени:
# fonlinebot/config.py
#...
MINUTE = 60
YEAR = 60*60*24*366
И допишем текста для блока настроек:
# fonlinebot/app/dialogs.py
# ...
league_row: str = "{i}. {flag} {name}"
config: str = "Сейчас выбраны:\n{leagues}"
btn_back: str = "<- Назад"
btn_go: str = "Вперед ->"
btn_save: str = "Сохранить"
config_btn_edit: str = "Изменить"
config_btn_delete: str = "Удалить данные"
data_delete: str = "Данные успешно удалены"
set_leagues: str = "Выбери 3 лиги для отслеживания.\nВыбраны:\n{leagues}"
main: str = "Что будем делать?"
db_saved: str = "Настройки сохранены"
cb_not_saved: str = "Лиги не выбраны"
cb_limit: str = "Превышен лимит. Максимум 3 лиги."
# ...
На этом подготовка окончена, пора написать логику добавления лиг. В модуль «service» добавим две inline-клавиатуры.
Inline-клавиатура привязана к конкретному сообщению. С их помощью можно отправлять и обрабатывать сигналы боту.
# 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
# ...
В клавиатуре CONFIG_KB 3 кнопки. Класс кнопки InlineKeyboardButton принимает текст и параметр callback_data. Именно callback нам отправит Телеграм после нажатия на кнопку. Вот так выглядит эта клавиатура:
А leagues_kb генерирует более сложную клавиатуру, с пагинацией. Мы выводим 5 лиг, кнопку «далее/назад» и «сохранить». Функция принимает выбранные лиги и отступ. Отступ нужен, для вывода лиг постранично.
Когда юзер нажимает на лигу, мы добавляем ее в кеш, нажимает повторно — удаляем. Обратите внимание, я генерирую динамическую строку в callback_data. Подставляю параметры offset и lg_id, что бы использовать при обработке.
Теперь напишем функцию красивого вывода списка лиг и обновления этого списка в кеше:
# fonlinebot/app/service.py
# ...
async def get_league_names(ids: list) -> str:
"""Функция собирает сообщение с названиями лиг из 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):
"""Функция добавляет или удаляет id лиги для юзера"""
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)
Настройка общения с Телеграмом
В файле бота допишем функции для меню настроек.
# fonlinebot/app/bot.py
@dp.callback_query_handler(lambda c: c.data == 'main_window')
async def show_main_window(callback_query: types.CallbackQuery):
"""Главный экран"""
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):
"""Обработка кнопки Настройки.
Проверка выбора лиг. Вывод меню изменений настроек"""
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):
"""Удаление пользователя из базы данных"""
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)
Функция get_config вызывается после нажатия на «Настройки». Если у пользователя выбраны лиги, она возвращает стандартное сообщение и меню настроек. В другом случае будет вызвана set_or_update_config для выбора лиг.
Одновременно я добавил удаление данных и главный экран.
Создавать и редактировать список будем в одной функции. Ей понадобятся параметры из строки callback. Например, пользователь нажал «вперед ->» и Телеграм прислал "edit_config#5". Мы разделили строку по # и взяли последнее значение (‘5’). Так будут передаваться параметры между сообщениями.
# 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=""):
"""Получение или обновление выбранных лиг"""
# если пришел callback, получим данные
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 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))
)
Не хватает еще реакции на нажатие по названию лиг и кнопке «сохранить».
# 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):
"""Добавление/удаление лиги из кеша, обновление сообщения"""
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):
"""Сохранение пользователя в базу данных"""
leagues_list = await s.get_league_ids(callback_query.from_user.id)
if len(leagues_list) > 3:
# не сохраняем, если превышен лимит лиг
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:
# не сохраняем если список пустой
await callback_query.answer(msg.cb_not_saved)
Готово. Теперь можете запустить бота и добавить лиги в настройках.
Сообщение с результатами матчей
Так как работа с API еще не настроена, напишем заглушку для этих процессов.
# 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):
"""Обработка команды online и кнопки Онлайн.
Запрос матчей. Вывод результатов"""
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):
"""Обновление сообщения результатов"""
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}"))
)
# игнорируем обновление, если прошло меньше минуты
await callback_query.answer(msg.cb_updated)
get_results получает лиги и фиксирует id сообщения для редактирования. Если у пользователя нет сохраненных лиг — вызывает set_or_update_config, в другом случает генерируем ответ с матчами и кнопкой «обновить».
Функция обновления результатов выполняет логику только если прошло больше минуты с момента последнего обновления. Для этого после каждого получения ответа с результатами мы добавляем запись в кеш со сроком хранения 1 минута.
Допишем зависимости:
# fonlinebot/app/dialogs.py
#...
results: str = "Все результаты за сегодня\n{matches}"
no_results: str = "Сегодня нет матчей"
update_results: str = "Обновить результаты"
cb_updated: str = f"{emojize(':white_heavy_check_mark:')} Готово"
#...
Кнопку обновления и генерацию ответа в «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:
"""Функция создaет сообщение для вывода результатов матчей"""
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:
"""Стандартизация ключей для хранения матчей"""
ids.sort()
return ",".join(ids)
async def parse_matches(ids: list) -> list:
"""Функция получения матчей по API"""
# логику напишем в следующей части
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:
# добавляем новые матчи, если они есть
cache.jset(lg_key, last_results, MINUTE)
return last_results
def results_to_text(matches: list) -> str:
"""
Функция генерации сообщения с матчами
"""
# логику напишем в следующей части
...
#...
Функция generate_results_answer получает матчи, преобразовывает данные в текст и возвращает его. Если матчей нет, возвращает соответствующий текст.
Что бы сэкономить ресурсы мы проверяем наличие матчей в кеше и только потом обращаемся к API.
Обработка неизвестных сообщений
Люди будут писать текст, который мы не обрабатываем. Обновите класс Messages:
# fonlinebot/app/dialogs.py
#...
unknown_text: str = "Ничего не понятно, но очень интересно.\nПопробуй команду /help"
#...
И unknown_message.
# fonlinebot/app/bot.py
#...
@dp.message_handler()
async def unknown_message(message: types.Message):
"""Ответ на любое неожидаемое сообщение"""
await message.answer(msg.unknown_text, reply_markup=s.MAIN_KB)
#...
Убедимся, что все работает:
Отлично. Мы написали бота с клавиатурами, кешированием и базой данных. Теперь пора добавить тестов.
Обновление тестов
Добавим класс TestService для контроля функции get_league_ids и get_last_results.
# 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)
#...
Ссылка на репозиторий с кодом в начале статьи. Удачи!
Что дальше?
Осталась малая часть работы. В следующей части мы найдем подходящий API и настроем обработку получения результатов: добавление внешнего API.
