Как-то я описывал создание бота для vdsina, чтобы мочь смотреть основные вещи быстро и удобно. В той статье я использовал библиотеку pyTelegramBotAPI для создания бота, сегодня посмотрим как тоже самое можно реализовать с помощью aiogram.
В прошлый раз мы использовали .env файл, давайте уйдем от него в сторону старых добрых .ini конфигурационных файлов.
[vdsina]
VDSINA_API_TOKEN = <token>
[telegram]
BOT_TOKEN = <token>
CHAT_ID = <allowed_chat_id>
[logger]
BOT_LOGGER_NAME = vdsina
BOT_LOGFILE = /var/log/vdsina.logНаш вспомогательный код меняется незначительно, в файле api.py, там где мы получали массив серверов, теперь будет получать мапу с серверами, это нам понадобится в дальнейшем, чтобы генерировать inline клавиатуру.
def get_servers_name(self):
servers_map = {}
response = self.send_api_requests(method="GET", url=f"{self.basic_url}/{self.servers_ep}")
try:
for _, value in enumerate(response["data"]):
servers_map[value["name"]] = value["name"]
except (KeyError, IndexError, TypeError) as err:
self.logger.error("Методо get_server_name() вернул ошибку: %s", err)
servers_map = None
return servers_mapНаш основной файл main.py теперь будет выглядеть так:
from logger import Logging
from api import basicApiCall, servers, account
import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters.command import Command
from aiogram.utils.keyboard import InlineKeyboardBuilder
from aiogram import F
import configparser
BOT_CONFIG = "/opt/vdsina/bot_config.ini"
config_parser = configparser.ConfigParser()
config_parser.read(BOT_CONFIG)
VDSINA_API_TOKEN = config_parser.get("vdsina", "VDSINA_API_TOKEN")
BOT_TOKEN = config_parser.get("telegram", "BOT_TOKEN")
ALLOW_CHAT_ID = config_parser.get("telegram", "CHAT_ID")
BOT_LOGGER_NAME = config_parser.get("logger", "BOT_LOGGER_NAME")
BOT_LOGFILE = config_parser.get("logger", "BOT_LOGFILE")
bot = Bot(BOT_TOKEN)
logger = Logging(BOT_LOGGER_NAME, BOT_LOGFILE)
api_instance = basicApiCall(VDSINA_API_TOKEN, logger)
servers_api = servers(api_instance.hoster_token, api_instance.logger)
account_api = account(api_instance.hoster_token, api_instance.logger)
dp = Dispatcher()
def servers_callback_mapping(keyboard_data):
button_map = {}
for row in keyboard_data:
for button in row:
button_map[button.callback_data] = button.text
return button_map
def builder_server_keyboard(servers, callback_data_prefix):
builder = InlineKeyboardBuilder()
for text, callback_data_name in servers.items():
callback_data_name = callback_data_name.replace("-", "_")
builder.add(
types.InlineKeyboardButton(text=text, callback_data=f"{callback_data_prefix}_{callback_data_name}")
)
return builder
@dp.message(Command("start"))
async def start(message: types.Message):
builder = InlineKeyboardBuilder()
builder.row(
types.InlineKeyboardButton(text="Проверить баланс", callback_data="vdsina_main_balance_status"),
types.InlineKeyboardButton(text="Прогноз отключения", callback_data="vdsina_main_account_data")
)
builder.row(
types.InlineKeyboardButton(text="Мои сервера", callback_data="vdsina_main_servers_params"),
types.InlineKeyboardButton(text="Мониторинг серверов", callback_data="vdsina_main_servers_stats")
)
builder.row(
types.InlineKeyboardButton(text="Помощь", callback_data="help")
)
await message.answer(
"Выберите что вас интересует из меню ниже:",
reply_markup=builder.as_markup()
)
@dp.message(Command("balance"))
async def balance_status(message: types.Message):
if message.chat.id == int(ALLOW_CHAT_ID):
balance_message = account_api.get_balance()
await message.answer(balance_message)
logger.info("Запрос статуса в чате: %s", message.chat.id)
else:
await message.answer("Это операция не разрешена для этого чата")
@dp.message(Command("account_forecast"))
async def account_forecast(message: types.Message):
if message.chat.id == int(ALLOW_CHAT_ID):
account_message = account_api.get_forecast()
await message.answer(account_message)
logger.info("Запрос прогноза отключения в чате: %s", message.chat.id)
else:
await message.answer("Это операция не разрешена для этого чата")
@dp.callback_query(F.data == "help")
async def help(callback: types.CallbackQuery):
await callback.message.answer("Для началы работы с ботом напишите /start")
@dp.callback_query(F.data.startswith("vdsina_main"))
async def vdsina_main_callback_process(callback: types.CallbackQuery):
if callback.message.chat.id == int(ALLOW_CHAT_ID):
if callback.data == "vdsina_main_balance_status":
balance_message = account_api.get_balance()
await callback.message.answer(balance_message)
logger.info("Запрос статуса в чате: %s", callback.message.chat.id)
elif callback.data == "vdsina_main_account_data":
account_message = account_api.get_forecast()
await callback.message.answer(account_message)
logger.info("Запрос прогноза отключения в чате: %s", callback.message.chat.id)
elif callback.data == "vdsina_main_servers_params":
servers = servers_api.get_servers_name()
builder = builder_server_keyboard(servers, "vdsina_servers_params")
await callback.message.answer("Выберите сервер:", reply_markup=builder.as_markup())
elif callback.data == "vdsina_main_servers_stats":
servers = servers_api.get_servers_name()
builder = builder_server_keyboard(servers, "vdsina_servers_stats")
await callback.message.answer("Выберите сервер:", reply_markup=builder.as_markup())
else:
await callback.message.answer("Это операция не разрешена для этого чата")
@dp.callback_query(F.data.startswith("vdsina_servers_params"))
async def vdsina_servers_params_callback_process(callback: types.CallbackQuery):
button_map = servers_callback_mapping(callback.message.reply_markup.inline_keyboard)
message = servers_api.get_server_params(button_map[callback.data])
await callback.message.answer(message)
logger.info("Запрос параметров сервера: %s, в чате: %s", button_map[callback.data], callback.message.chat.id)
@dp.callback_query(F.data.startswith("vdsina_servers_stats"))
async def vdsina_servers_stats_callback_process(callback: types.CallbackQuery):
button_map = servers_callback_mapping(callback.message.reply_markup.inline_keyboard)
images_path = servers_api.get_server_monitoring(button_map[callback.data])
if images_path is not None:
for image in images_path:
caption = image.split("/")[1].replace(".png", "").upper()
await callback.message.answer_photo(types.FSInputFile(path=image), caption=caption)
logger.info("Запрос мониторинга сервера: %s, в чате: %s", button_map[callback.data], callback.message.chat.id)
else:
await callback.message.answer("Не удалось отправить графики")
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())Давайте разберем, что тут понаписано.
BOT_CONFIG = "/opt/vdsina/bot_config.ini"
config_parser = configparser.ConfigParser()
config_parser.read(BOT_CONFIG)
VDSINA_API_TOKEN = config_parser.get("vdsina", "VDSINA_API_TOKEN")
BOT_TOKEN = config_parser.get("telegram", "BOT_TOKEN")
ALLOW_CHAT_ID = config_parser.get("telegram", "CHAT_ID")
BOT_LOGGER_NAME = config_parser.get("logger", "BOT_LOGGER_NAME")
BOT_LOGFILE = config_parser.get("logger", "BOT_LOGFILE")
bot = Bot(BOT_TOKEN)
logger = Logging(BOT_LOGGER_NAME, BOT_LOGFILE)
api_instance = basicApiCall(VDSINA_API_TOKEN, logger)
servers_api = servers(api_instance.hoster_token, api_instance.logger)
account_api = account(api_instance.hoster_token, api_instance.logger)
dp = Dispatcher()Начала весьма стандартно, мы получаем переменные из конфиг файла и создаем экземпляры наших классов, они будут нужны для извлечения данных.
О функциях servers_callback_mapping() и builder_server_keyboard() поговорим чуть ниже, когда логически подойдем к ним.
Команда start:
@dp.message(Command("start"))
async def start(message: types.Message):
builder = InlineKeyboardBuilder()
builder.row(
types.InlineKeyboardButton(text="Проверить баланс", callback_data="vdsina_main_balance_status"),
types.InlineKeyboardButton(text="Прогноз отключения", callback_data="vdsina_main_account_data")
)
builder.row(
types.InlineKeyboardButton(text="Мои сервера", callback_data="vdsina_main_servers_params"),
types.InlineKeyboardButton(text="Мониторинг серверов", callback_data="vdsina_main_servers_stats")
)
builder.row(
types.InlineKeyboardButton(text="Помощь", callback_data="help")
)
await message.answer(
"Выберите что вас интересует из меню ниже:",
reply_markup=builder.as_markup()
)Здесь мы обрабатываем команду «/start«, создаем inline клавиатуры и возвращаем ее вместе с текстом «Выберите что вас интересует из меню ниже». Я не нашел способа, как можно было бы передать параметр row_width или что-то похожее, чтобы клавиатура приняла вид в две колонки. Поэтому пришлось отказаться от идее сделать переменную в виде массива и создавать клавиатуру из цикла.
Команда balance и account_forecast:
@dp.message(Command("balance"))
async def balance_status(message: types.Message):
if message.chat.id == int(ALLOW_CHAT_ID):
balance_message = account_api.get_balance()
await message.answer(balance_message)
logger.info("Запрос статуса в чате: %s", message.chat.id)
else:
await message.answer("Это операция не разрешена для этого чата")
@dp.message(Command("account_forecast"))
async def account_forecast(message: types.Message):
if message.chat.id == int(ALLOW_CHAT_ID):
account_message = account_api.get_forecast()
await message.answer(account_message)
logger.info("Запрос прогноза отключения в чате: %s", message.chat.id)
else:
await message.answer("Это операция не разрешена для этого чата")Две похожие команды, я их вынес в отдельные обработчики именно команд, чтобы не вызывать их каждый раз из меню.
Обработчик callback vdsina_main:
@dp.callback_query(F.data.startswith("vdsina_main"))
async def vdsina_main_callback_process(callback: types.CallbackQuery):
if callback.message.chat.id == int(ALLOW_CHAT_ID):
if callback.data == "vdsina_main_balance_status":
balance_message = account_api.get_balance()
await callback.message.answer(balance_message)
logger.info("Запрос статуса в чате: %s", callback.message.chat.id)
elif callback.data == "vdsina_main_account_data":
account_message = account_api.get_forecast()
await callback.message.answer(account_message)
logger.info("Запрос прогноза отключения в чате: %s", callback.message.chat.id)
elif callback.data == "vdsina_main_servers_params":
servers = servers_api.get_servers_name()
builder = builder_server_keyboard(servers, "vdsina_servers_params")
await callback.message.answer("Выберите сервер:", reply_markup=builder.as_markup())
elif callback.data == "vdsina_main_servers_stats":
servers = servers_api.get_servers_name()
builder = builder_server_keyboard(servers, "vdsina_servers_stats")
await callback.message.answer("Выберите сервер:", reply_markup=builder.as_markup())
else:
await callback.message.answer("Это операция не разрешена для этого чата")Не смог найти, как можно затащить все коллбеки и обрабатывать их if-ами, как я это делал в прошлый раз, поэтому решил сгруппировать имена коллбеков. Все что начинается с vdsina_main относится к основному меню.
Собственно тут все стандартно, за исключением двух последних коллбеков.
Я очень хотел уйти от старой реализации, где нужно вводить руками имя сервера к вот такому виду:
Вероятно есть более правильный способ реализации, но я сделал это просто через создание еще одной inline клавиатуры. Создается она в этом методе:
def builder_server_keyboard(servers, callback_data_prefix):
builder = InlineKeyboardBuilder()
for text, callback_data_name in servers.items():
callback_data_name = callback_data_name.replace("-", "_")
builder.add(
types.InlineKeyboardButton(text=text, callback_data=f"{callback_data_prefix}_{callback_data_name}")
)
return builderЗдесь мы также передаем нужный префикс, чтобы потом было проще работать с коллбеками, которые будут от этой клавиатуры.
Обработка коллбеков от имен серверов:
@dp.callback_query(F.data.startswith("vdsina_servers_params"))
async def vdsina_servers_params_callback_process(callback: types.CallbackQuery):
button_map = servers_callback_mapping(callback.message.reply_markup.inline_keyboard)
message = servers_api.get_server_params(button_map[callback.data])
await callback.message.answer(message)
logger.info("Запрос параметров сервера: %s, в чате: %s", button_map[callback.data], callback.message.chat.id)Тут обрабатываем все, что начинается с vdsina_servers_params. И тут возникает небольшой нюанс. Мы создали клавиатуру, которая нам возврщает имена серверов в виде кнопок, но как понять, что пользователь выбрал именно эту кнопку? Конечно, можно это сделать во callback имени. Но дабы не городить много if-ов там, где можно и без этого, я сделал метод с маппингом, возвращаемой клавиутуры:
def servers_callback_mapping(keyboard_data):
button_map = {}
for row in keyboard_data:
for button in row:
button_map[button.callback_data] = button.text
return button_mapВ обработчике коллбеков от vdsina_servers_params мы получаем сначала маппинг наших коллбек имен и текста, который написан на кнопки. Далее, чтобы получить имя сервера и передать его в экземпляр класса, мы передаем туда button_map[callback.data].
callback.data — это имя нашего коллбека, по его имени нам вернется текст кнопки, который соответствует нужному имени сервера. Тем самым данный обработчик получается универсальным. Добавив еще серверов, мы получим их на предыдущих шага, из их кол-ва сформируем клавиатуру и получим данные по серверу.
Примерно тоже самое с отправкой изображения с мониторингом:
@dp.callback_query(F.data.startswith("vdsina_servers_stats"))
async def vdsina_servers_stats_callback_process(callback: types.CallbackQuery):
button_map = servers_callback_mapping(callback.message.reply_markup.inline_keyboard)
images_path = servers_api.get_server_monitoring(button_map[callback.data])
if images_path is not None:
for image in images_path:
caption = image.split("/")[1].replace(".png", "").upper()
await callback.message.answer_photo(types.FSInputFile(path=image), caption=caption)
logger.info("Запрос мониторинга сервера: %s, в чате: %s", button_map[callback.data], callback.message.chat.id)
else:
await callback.message.answer("Не удалось отправить графики")Здесь я добавил подпись к каждой картинке.
Финальным штрихом меняем наш vdsina.service:
[Unit]
Description=Vdsina Bot
After=multi-user.target
[Service]
Type=simple
Restart=always
ExecStart=/usr/bin/python3 /opt/vdsina/main.py
[Install]
WantedBy=multi-user.targetПолный код можно найти тут.
