Есть у меня 3 небольших сервера, которые хостятся у VDSINA. Статья если что не заказная, хостер мне за нее ничего не платил, просто так совпало, что я размещаю машины у них и у них же есть API через который мы можем получать всякое разное по нашим серверам и аккаунту. А если это еще будет у нас в телеграмме, то будет достаточно удобно и не нужно постоянно заходить в админ панель.
Какие функции будет выполнять наш бот?
- Уметь показывать баланс по аккаунту.
- Уметь показывать прогноз отключение серверов.
- Выводить информацию по конфигурации наших серверов, с возможностью выбора конкретного сервера.
- Выводить информацию по основным показателям мониторинга, таких как IOPS по дискам, утилизация CPU в процентах, входящий и исходящий трафик, строить на основе этих данных график, конвертировать его в изображение и отправлять нам в телеграм.
ТЗ описано, давайте приступать.
Начинаем описание с main.py
import telebot
from telebot import types
from logger import Logging
from api import basicApiCall, servers, account
import os
import datetime
current_date = datetime.datetime.now().date()
BOT_TOKEN = os.environ.get("BOT_TOKEN")
VDSINA_API_TOKEN = os.environ.get("VDSINA_API_TOKEN")
LOG_FILE = f"/var/log/{current_date}_vdsina_bot.txt"
CHAT_ID = os.environ.get("CHAT_ID")
bot = telebot.TeleBot(BOT_TOKEN)
logger = Logging("vdsina_bot", LOG_FILE)
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)
@bot.message_handler(commands=["start"])
def start(msg):
markup_inline = types.InlineKeyboardMarkup(row_width=2)
balance_button = types.InlineKeyboardButton("Проверить баланс", callback_data="balance_status")
account_button = types.InlineKeyboardButton("Прогноз отключения", callback_data="account_data")
servers_params_button = types.InlineKeyboardButton("Мои сервера", callback_data="servers_params")
servers_stats_button = types.InlineKeyboardButton("Мониторинг серверов", callback_data="servers_stats")
help_button = types.InlineKeyboardButton("Помощь", callback_data="help")
markup_inline.add(balance_button, account_button, servers_params_button, servers_stats_button, help_button)
if msg.text == "/start":
bot.send_message(msg.chat.id, "Информационный серверный бот", reply_markup=markup_inline)
def main():
bot.polling(none_stop=True, interval=0)
if __name__ == "__main__":
main()Используемые глобальные переменные:
BOT_TOKEN — это токен, который мы получили при создании бота через другого бота BotFather в телеграм.
VDSINA_API_TOKEN — конкретно у этого хостера, токен можно получить в ЛК. Я думаю, что у любого другого также, если у них есть API.
LOG_FILE — переменная с полным путем для сохранения лога. Наш бот будет писать логи в json формате. Об этом будет ниже. Имя файла в качестве префикса будет иметь текущую дату в формате — год-месяц-день.
CHAT_ID — это переменная с ID нашего чата в телеграм, есть множество способов узнать его, через тоже API телеграм или с помощью лога этого бота, суть переменной — безопасность. Наш бот имеет доступ к нашим серверам, а следовательно любой, кто найдет его в телеграм, может получить информацию, которая ему не предназначалась, к тому же, в будущем возможно мы захотим расширить функционал бота и например научить его создавать новые сервера или перезагружать старые, все это мы должны мочь делать только с нашего аккаунта, но никак из другого.
Далее мы создаем экземпляры наших классов, о них будет чуть ниже.
После этого через декоратор создаем меню для бота и возвращаем это меню, если пользователь вводит «/start».
Уже сейчас наш бот будет выглядить примерно вот так:
Далее пишем по сути «роутинг» для этих кнопок.
@bot.callback_query_handler(func=lambda call:True)
def callback(call):
if call.message and call.message.chat.id == int(CHAT_ID):
if call.data == "help":
bot.send_message(call.message.chat.id, "Для началы работы с ботом напишите /start")
elif call.data == "balance_status":
message = account_api.get_balance()
bot.send_message(call.message.chat.id, message)
logger.info("Запрос статуса в чате: %s", call.message.chat.id)
elif call.data == "account_data":
message = account_api.get_forecast()
bot.send_message(call.message.chat.id, message)
logger.info("Запрос прогноза отключения в чате: %s", call.message.chat.id)
elif call.data == "servers_params":
servers = servers_api.get_servers_name()
if servers is not None:
servers_str = ' | '.join(servers)
msg = bot.send_message(call.message.chat.id, servers_str)
msg = bot.send_message(call.message.chat.id, "Введите имя сервера: ")
bot.register_next_step_handler(msg, get_servers_params, servers)
else:
msg = bot.send_message(call.message.chat.id, "Пустой массив servers")
elif call.data == "servers_stats":
servers = servers_api.get_servers_name()
if servers is not None:
servers_str = ' | '.join(servers)
msg = bot.send_message(call.message.chat.id, servers_str)
msg = bot.send_message(call.message.chat.id, "Введите имя сервера: ")
bot.register_next_step_handler(msg, get_servers_monitoring, servers_str, call.message.chat.id)
else:
msg = bot.send_message(call.message.chat.id, "Пустой массив servers")
else:
bot.send_message(call.message.chat.id, "Это операция не разрешена для этого чата.")Здесь мы уже видим, зачем нам была нужна переменная CHAT_ID. Теперь, если «левый чел» захочет что-то посмотреть через нашего бота, то выглядеть для него это будет примерно так:
Давайте пробежимся по основным «роутам» здесь:
Помощь:
if call.data == "help":
bot.send_message(call.message.chat.id, "Для началы работы с ботом напишите /start")Все предельно просто, если пользователь нажимает кнопку «Помощь», то ему в ответ будет направлено сообщение через метод send_message.
Проверить баланс:
elif call.data == "balance_status":
message = account_api.get_balance()
bot.send_message(call.message.chat.id, message)
logger.info("Запрос статуса в чате: %s", call.message.chat.id)Тут мы в переменную message записываем некое значение, которое получаем через метод экземпляра класса account_api. Сейчас мы не знаем какая там логика, об этом я расскажу ниже. На данный момент достаточно того, что есть некий метод, который вернет нам сообщение, как он реализован нас пока не интересует.
Прогноз отключения:
elif call.data == "account_data":
message = account_api.get_forecast()
bot.send_message(call.message.chat.id, message)
logger.info("Запрос прогноза отключения в чате: %s", call.message.chat.id)Абсолютно аналогичная ситуация. Получаем сообщение из некоего пока не известного нам метода.
Мои сервера:
elif call.data == "servers_params":
servers = servers_api.get_servers_name()
if servers is not None:
servers_str = ' | '.join(servers)
msg = bot.send_message(call.message.chat.id, servers_str)
msg = bot.send_message(call.message.chat.id, "Введите имя сервера: ")
bot.register_next_step_handler(msg, get_servers_params, servers)
else:
msg = bot.send_message(call.message.chat.id, "Пустой массив servers")Тут уже сложнее и интереснее. Мы получаем переменную servers через метод экземпляра класса servers_api, забегая вперед скажу, что это должен быть массив со списком наших серверов. Далее мы проверяем что массив не равен None, если это верно, то мы преобразовываем массив в строку, чтобы она приняла вид «server1 | server2». Далее мы отправляем сообщение в чат со списком серверов. Преобразование в строку было нужно как раз для этого, иначе отправился только бы первый элемент массива. Далее мы отправляем еще одно сообщение, с предложение ввести имя сервера, после чего мы говорим боту, что обработка ответа пользователя уходит в метод get_servers_params. О нем будет ниже.
Также, если по каким-то причинам мы не смогли получить список наших серверов, бот отправит нам сообщение об этом.
Мониторинг серверов:
elif call.data == "servers_stats":
servers = servers_api.get_servers_name()
if servers is not None:
servers_str = ' | '.join(servers)
msg = bot.send_message(call.message.chat.id, servers_str)
msg = bot.send_message(call.message.chat.id, "Введите имя сервера: ")
bot.register_next_step_handler(msg, get_servers_monitoring, servers_str, call.message.chat.id)
else:
msg = bot.send_message(call.message.chat.id, "Пустой массив servers")Здесь логика точно такая же как и в «Моих серверах».
Окей тут разобрались, двигаемся дальше к оставшейся части кода файла main.py.
@bot.message_handler(content_types=["text"])
def message_handler(m):
bot.send_message(m.chat.id, "Неверная команда")
def get_servers_params(msg, servers):
if msg.text in servers:
message = servers_api.get_server_params(msg.text)
bot.send_message(msg.chat.id, message)
logger.info("Запрос параметров сервера: %s, в чате: %s", msg.text, msg.chat.id)
else:
bot.send_message(msg.chat.id, "Выбранный сервер не существует")
logger.warning("Введено неправильное имя сервера: %s, в чате: %s", msg.text, msg.chat.id)
def get_servers_monitoring(msg, servers, chat_id):
if msg.text in servers:
images_path = servers_api.get_server_monitoring(msg.text)
if images_path is not None:
for image in images_path:
with open(image, 'rb') as img:
bot.send_photo(chat_id, img)
logger.info("Запрос мониторинга сервера: %s", msg.text)
else:
bot.send_message(chat_id, "Не удалось отправить графики")
else:
bot.send_message(msg.chat.id, "Выбранный сервер не существует")
logger.warning("Введено неправильное имя сервера: %s, в чате: %s", msg.text, msg.chat.id)Давайте рассмотрим вот этот кусок:
@bot.message_handler(content_types=["text"])
def message_handler(m):
bot.send_message(m.chat.id, "Неверная команда")Задумка в том, что любая команда отличная от «/start» должна игнорироваться ботом, а лучше обрабатываться как здесь (у нас же кнопки, помните). Поэтому если просто писать боту что-то, он будет отправлять «Неверная команда».
Метод get_servers_params:
def get_servers_params(msg, servers):
if msg.text in servers:
message = servers_api.get_server_params(msg.text)
bot.send_message(msg.chat.id, message)
logger.info("Запрос параметров сервера: %s, в чате: %s", msg.text, msg.chat.id)
else:
bot.send_message(msg.chat.id, "Выбранный сервер не существует")
logger.warning("Введено неправильное имя сервера: %s, в чате: %s", msg.text, msg.chat.id)Один из методов куда мы «роутим» нашу кнопку, точнее ее backend. Здесь в методе мы принимаем сообщение, в данном случае в переменной msg содержится помимо сообщения, различная служебная информация от телеграм, например chat_id и прочее и список серверов в переменной servers. Напомню, что их мы передали выше, в качестве аргумента метода.
Проверяем что сообщение, которое мы ввели, соответствует тем именам серверов, которые в нашем списке в переменной servers. Далее как уже было выше, мы записываем в переменную message что-то, что получаем из другого класса и отправляем это в чат.
Метод get_servers_monitoring:
def get_servers_monitoring(msg, servers, chat_id):
if msg.text in servers:
images_path = servers_api.get_server_monitoring(msg.text)
if images_path is not None:
for image in images_path:
with open(image, 'rb') as img:
bot.send_photo(chat_id, img)
logger.info("Запрос мониторинга сервера: %s", msg.text)
else:
bot.send_message(chat_id, "Не удалось отправить графики")
else:
bot.send_message(msg.chat.id, "Выбранный сервер не существует")
logger.warning("Введено неправильное имя сервера: %s, в чате: %s", msg.text, msg.chat.id)Этот метож похож на предыдущий и использует теже парадигмы, за исключением того, что здесь мы получем в переменной images_path пути до сгенерированных изображений с графиками и отправляем не сообщение в виде текса, а именно картинки нам в чат.
Мы разобрали основной файл main.py, давайте перейдем к следующему — api.py:
from dataclasses import dataclass
import requests
from logger import Logging
from graph import graphProcess
import os
@dataclass
class basicApiCall:
hoster_token: str
logger: Logging
basic_url: str = "https://userapi.vdsina.ru"
balance_ep: str = "v1/account.balance"
account_ep: str = "v1/account"
servers_ep: str = "v1/server"
servers_mon_ep: str = "v1/server.stat"
images_dir: str = "images"Импортируем нужные нам пакеты и наш другой класс graphProcess. Тут мы используем декоратор @dataclass, он автоматически за нас создаст ряд магических методов, в том числе метод __init__, следовательно нам достаточно только описать поля, нет необходимости создавать конструктор класса самим.
Поля:
hoster_token — наш токен для хостинга.
logger — принимает сам класс, что делает переменную экземпляром класса, теперь мы можем использовать это для обращения к методам класса Logging.
Все остальные поля заданы с дефолтными значениями, думаю из их названия понятно для чего они служат.
Разбираем методы класса basicApiCall:
def send_api_requests(self, **kwargs):
headers = {
'Authorization': f'{self.hoster_token}',
'Content-Type': 'application/json'
}
if kwargs.get("method") == "GET":
try:
response = requests.get(url=kwargs.get("url"), headers=headers)
if response.status_code == 200:
return response.json()
except requests.exceptions.RequestException as err:
self.logger.warning("Ошибка при отправке запроса на %s. Код ответа %s", kwargs.get("url"), response.status_code)
return NoneМетод для отправки request запроса с различным типом, на данный момент реалилован только GET.
def get_servers_name(self):
servers_list = []
response = self.send_api_requests(method="GET", url=f"{self.basic_url}/{self.servers_ep}")
try:
for _, value in enumerate(response["data"]):
servers_list.append(value["name"])
except (KeyError, IndexError, TypeError) as err:
self.logger.error("Методо get_server_name() вернул ошибку: %s", err)
servers_list = None
return servers_listМы уже видили этот метод из описание файла main.py. Тут все просто, мы отправляем API запрос на все доступные нам сервера, получаем ответ в виде словаря, проходимся по нему циклом и получаем только имена наших серверов, их записываем в массив servers_list. Тут же встроена обработка исключений, если мы получим одну из возможных ошибок при работае со списками, словарями, то присвоим переменной servers_list значение None. А уж то, как обработается такое значение я уже описал выше.
def make_servers_mapping(self):
servers_dict = {}
response = self.send_api_requests(method="GET", url=f"{self.basic_url}/{self.servers_ep}")
try:
for _, value in enumerate(response["data"]):
servers_dict[value["name"]] = value["id"]
except (KeyError, IndexError, TypeError) as err:
self.logger.error("Методо make_servers_mapping() вернул ошибку: %s", err)
servers_dict = None
return servers_dictСуть метода сопоставить имя сервера с его id. Так как мы не можем получить параметры конкретного сервера по имени через API, нам нужно отправлять запрос на ID. В этом методе мы матчим имена и «айдишники».
Разбираем методы класса servers:
@dataclass
class servers(basicApiCall):
def get_server_params(self, server_name):
servers_mapping = self.make_servers_mapping()
response = self.send_api_requests(method="GET", url=f"{self.basic_url}/{self.servers_ep}/{servers_mapping[server_name]}")
try:
server_status = response["status"]
server_ip_addr = response["data"]["ip"][0]["ip"]
cpu_count = response["data"]["data"]["cpu"]["value"]
ram_count = response["data"]["server-plan"]["name"]
disk_volume = response["data"]["data"]["disk"]["value"]
disk_calculation = response["data"]["data"]["disk"]["for"]
traff_volume = response["data"]["data"]["traff"]["value"]
traff_calculation = response["data"]["data"]["traff"]["for"]
os_release = response["data"]["template"]["name"]
location = response["data"]["datacenter"]["name"]
message = (
f"Имя сервера: {server_name}\n"
f"Статус: {server_status}\n"
f"ОС: {os_release}\n"
f"IP адрес: {server_ip_addr}\n"
f"Кол-во ЦПУ: {cpu_count}\n"
f"RAM: {ram_count}\n"
f"Объем диска: {disk_volume}{disk_calculation}\n"
f"Объем трафика: {traff_volume}{traff_calculation}\n"
f"Датацентр: {location}"
)
except (KeyError, IndexError, TypeError) as err:
self.logger.error("Метод get_server_params() вернул ошибку: %s", err)
message = f"Ошибка при получении параметров сервера: {err}"
return messageВ телеграме это выглядит так:
Этот класс наследуется от базовго. По сути здесь мы собираем нужные нам данные о нашем сервере и формируем сообщение. Думаю остальное вполне понятно.
def get_server_monitoring(self, server_name):
servers_mapping = self.make_servers_mapping()
response = self.send_api_requests(method="GET", url=f"{self.basic_url}/{self.servers_mon_ep}/{servers_mapping[server_name]}")
try:
response = response["data"][-3:]
graphProcess(response).make_graph_images()
image_files = os.listdir(self.images_dir)
image_files = [filename for filename in image_files if filename.lower().endswith(('.jpg', '.jpeg', '.png', '.gif'))]
image_files_with_path = [f"images/{filename}" for filename in image_files]
except Exception as err:
self.logger.error("Метод get_server_monitoring() вернул ошибку: %s", err)
image_files_with_path = None
return image_files_with_pathЭто самый интересный метод, потому что здесь присутствует вот эта строка graphProcess(response).make_graph_images(). Именно там происходит построение графиком и их сохранение в виде изображений. Но об этом будет чуть позже. Пока достаточно знать следующие:
API отдаем нам мониторинг в виде огромного json файла, где внутри ключа data находится список из элементов с информацией от «дефолтного» мониторинга хостера, где каждый элемент это список, в котором есть ключ dt — это дата увеличивающаяся ровно на один час. Общий объем данных равен суткам, которые поделены на каждый час. Те в конечном счете мы видем некий временной ряд с хопом в час.
response = response[«data»][-3:] — здесь мы берем последние три элемента списка, те последние 3 часа от текущего времени.
Далее мы формируем список с полученными в методе graphProcess(response).make_graph_images() изображениями.
Выглядеть это будет так:
Разбираем методы класса account:
@dataclass
class account(basicApiCall):
def get_balance(self):
response = self.send_api_requests(method="GET", url=f"{self.basic_url}/{self.balance_ep}")
try:
status = response["status"]
balance_real = response["data"]["real"]
balance_bonus = response["data"]["bonus"]
balance_partner = response["data"]["partner"]
message = (
f"Статус: {status}\n"
f"Баланс: {balance_real} руб\n"
f"Бонусы: {balance_bonus} руб\n"
f"Партеры: {balance_partner} руб"
)
except (KeyError, IndexError, TypeError) as err:
self.logger.error("Метод get_balance() вернул ошибку: %s", err)
message = f"Ошибка при получении баланса: {err}"
return message
def get_forecast(self):
response = self.send_api_requests(method="GET", url=f"{self.basic_url}/{self.account_ep}")
try:
forecast = response["data"]["forecast"]
message = (
"Формат даты: гг-мм-дд\n"
f"Будет отключено: {forecast}"
)
except (KeyError, IndexError, TypeError) as err:
self.logger.error("Метод get_forecast() вернул ошибку: %s", err)
message = f"Ошибка при получении прогноза отключения: {err}"
return messageТут нет ничего нового, что бы мы уже не видили ранее, я думаю что из кода достаточно просто понять, что делает каждый метод.
Смотрится так:
Класс для построение графиков:
import matplotlib
import matplotlib.pyplot as plt
from datetime import datetime
from dataclasses import dataclass
import os
@dataclass
class graphProcess:
data: dict
images_dir: str = "images"
def make_graph_images(self):
## Включаем не интерактивный режим
matplotlib.use('Agg')
timestamps = []
cpu_values = []
disk_writes = []
disk_reads = []
vnet_tx = []
vnet_rx = []
time_interval_seconds = 3600
if not os.path.exists(self.images_dir):
os.makedirs(self.images_dir)
for entry in self.data:
dt = datetime.strptime(entry["dt"], "%Y-%m-%d %H:%M:%S")
timestamps.append(dt)
cpu_values.append(entry["stat"]["cpu"])
disk_writes.append(entry["stat"]["disk_writes"] / time_interval_seconds)
disk_reads.append(entry["stat"]["disk_reads"] / time_interval_seconds)
vnet_tx_kbps = (entry["stat"]["vnet_tx"] * 8) / 1024
vnet_tx_kbps = vnet_tx_kbps / time_interval_seconds
vnet_rx_kbps = (entry["stat"]["vnet_rx"] * 8) / 1024
vnet_rx_kbps = vnet_rx_kbps / time_interval_seconds
vnet_tx.append(vnet_tx_kbps)
vnet_rx.append(vnet_rx_kbps)
## Создаем график для CPU
fig1, ax1 = plt.subplots()
ax1.plot(timestamps, cpu_values, color='tab:blue')
ax1.set_xlabel('Timestamp')
ax1.set_ylabel('CPU (%)')
ax1.set_title('CPU Usage')
## Создаем график для IOPS
fig2, ax2 = plt.subplots()
ax2.plot(timestamps, disk_reads, color='tab:purple', label='Disk Reads')
ax2.plot(timestamps, disk_writes, color='tab:green', label='Disk Writes')
ax2.set_xlabel('Timestamp')
ax2.set_ylabel('Disk Operations (IOPS)')
ax2.set_title('Disk Operations')
ax2.legend()
## Создаем график для VNET
fig3, ax3 = plt.subplots()
ax3.plot(timestamps, vnet_rx, color='tab:blue', label='VNET RX')
ax3.plot(timestamps, vnet_tx, color='tab:orange', label='VNET TX')
ax3.set_xlabel('Timestamp')
ax3.set_ylabel('VNET Operations (Kbps)')
ax3.set_title('VNET Operations')
ax3.legend()
## Сохраняем график в изображение
fig1.savefig(f"{self.images_dir}/cpu_usage.png")
fig2.savefig(f"{self.images_dir}/disk.png")
fig3.savefig(f"{self.images_dir}/vnet.png")
## Закрываем фигуры
plt.close(fig1)
plt.close(fig2)
plt.close(fig3)Комментарии в коде, в целом интерес для нас представляет только это
matplotlib.use(‘Agg’)
Без использования этого параметра я ловил ошибку
Она возникает из-за того, что я вызываю метод «subplots» в не основном классе, параметр выше отключает использование интерактивного режима и фиксит эту ошибку.
disk_writes.append(entry[«stat»][«disk_writes»] / time_interval_seconds) — iops мы получаем так, потому что интервал наших данных составляет один час.
Аналогично и для переменных с трафиком, только там мы получаем значения в байтах, а хотим видеть в килобитах.
1 килобит (kb) = 1024 бита (bits)
1 байт (byte) = 8 бит (bits)
Следовательно получаем формулы — Килобиты = (Байты × 8) / 1024 и так как мы хотим видеть верную динамику за наш период времени, то мы делим полученное значение на один час.
Логирование:
import logging
import logging.handlers
import json
class JSONFormatter(logging.Formatter):
def format(self, record):
log_record = {
'asctime': self.formatTime(record, self.datefmt),
'filename': record.filename,
'codeLine': record.lineno,
'levelname': record.levelname,
'message': record.getMessage(),
}
return json.dumps(log_record, ensure_ascii=False)
class Logging():
def __init__(self, logger_name, log_file):
self.logger_name = logger_name
self.logger = logging.getLogger(self.logger_name)
self.logger.setLevel(logging.INFO)
formatter = JSONFormatter()
file_handler = logging.handlers.TimedRotatingFileHandler(
log_file, when='midnight', interval=1, backupCount=7
)
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(formatter)
self.logger.addHandler(file_handler)
def log(self, level, message, *args):
self.logger.log(level, message, *args)
def info(self, message, *args):
self.log(logging.INFO, message, *args)
def debug(self, message, *args):
self.log(logging.DEBUG, message, *args)
def warning(self, message, *args):
self.log(logging.WARNING, message, *args)
def error(self, message, *args):
self.log(logging.ERROR, message, *args)
def critical(self, message, *args):
self.log(logging.CRITICAL, message, *args)Тут мы по сути реализуем свой класс для логирование, добавляем кастомный форматтер, который будет нам писать красивые логи в json и добавляем автоматическую ротацию логов, старше 7 дней, а также возможность писать в новый файл, когда стрелка на часах будет больше 00:00 часов. Все это тут
file_handler = logging.handlers.TimedRotatingFileHandler(
log_file, when='midnight', interval=1, backupCount=7
)Логи будут выглядеть следующим образом:
Последние штрихи:
Для работы кода нужно поставить две библиотеки через pip
pyTelegramBotAPI
matplotlibТочно будет работать на python3.8 и выше, все что ниже, возможны различные спецэффекты)
Создадим systemd service:
root@v2082402:/home/admin# cat /etc/systemd/system/vdsina_bot.service
[Unit]
Description=Telegram bot
After=multi-user.target
[Service]
EnvironmentFile=/opt/bot/.env
Type=simple
Restart=always
ExecStart=/usr/bin/python3 /opt/bot/main.py
[Install]
WantedBy=multi-user.targetEnvironmentFile=/opt/bot/.env — так как systemd запускаем процессы в пространстве отличным от шелла вашего юзера, то переменные окружения, которые мы могли бы определить например в .bashrc работать не будут. Поэтому создадим .env в директории с ботом. Это не лучший способ но у нас нет VAULT, поэтому пока так.
На этом все, также полный код бота можно найти на моем github.
