Всем привет! Сегодня я расскажу, как создал Telegram-бота, который помогает мне следить за сервером и выполнять команды прямо из чата.
Начнем с задачи.
Нужно сделать бота, который будет жить на сервере, мониторить состояние служб и выполнять команды через Telegram-интерфейс.
Функционал бота:
- Мониторит состояние сервера и отправляет уведомления в Telegram.
- Выполняет команды на сервере через Telegram-интерфейс.
- Настраивается через конфигурационный файл, где задаются команды и параметры уведомлений.
- Ведет логи своих действий (отправка уведомлений, выполнение команд).
Начнем с того, что определим как бот будет принимать сообщения и как будут настраиваться команды, которые нужно выполнять на сервере.
Формат сообщения, на первое время, выбрал следующий:
{
"status": "alert",
"service": "network",
"message": "interface eth0 is down"
}
Сообщения будут приходить в формате JSON, что позволяет легко передавать данные о состоянии сервера.
Команды будут задаваться в конфигурационном файле, что делает бота гибким и легко настраиваемым.
Пример конфига:
{
"server_info":
{
"server_name":"hostname'",
"server_ip":"curl -s ifconfig.me",
"CPU":"cat /proc/loadavg | awk '{print $1}'",
"RAM":"free -m | awk 'NR==2 {print $2 \"/\" $3}'"
},
"server_command_menu":
{
"Состояние бота 1":"status_bot1",
"Состояние бота 2":"status_bot2",
"Состояние бота 3":"status_bot3"
},
"status_bot1":"systemctl status bot1.service | grep 'Active'",
"status_bot2":"systemctl status bot2.service | grep 'Active'",
"status_bot3":"systemctl status bot3.service | grep 'Active'"
}
Для того чтобы бот принимал сообщения о различных ситуациях на сервере и присылал в тг, буду использовать простой клиент-сервер на сокетах.
Вот пример серверной части (запускается с ботом):
HOST = '0.0.0.0'
PORT = 12345
async def start_server_agent(bot:Bot):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
server_socket.bind((HOST, PORT))
server_socket.listen()
server_socket.settimeout(1)
logger_server_agent.info(f"Сервер слушает {HOST}:{PORT}...")
try:
while True:
try:
conn, addr = server_socket.accept()
except socket.timeout:
await asyncio.sleep(1)
continue
with conn:
logger_server_agent.info(f"Подключение от {addr}")
data = conn.recv(1024)
if not data:
continue
try:
message = json.loads(data.decode("utf-8"))
formatted_msg = f"🔹 *Service Alert*\n📌 {message.get('service', 'Unknown')}\n⚠ {message.get('message', 'No details')}"
await bot.send_message(settings.bots.admin_id, formatted_msg, parse_mode="Markdown")
except json.JSONDecodeError:
logger_server_agent.error("Ошибка декодирования JSON")
except asyncio.CancelledError:
logger_server_agent.info("Сервер остановлен.")
А вот простой клиент для теста:
SERVER_IP = '127.0.0.1'
PORT = 12345
message = {
"status": "alert",
"service": "network",
"message": "interface eth0 is down"
}
def send_message():
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client_socket:
client_socket.connect((SERVER_IP, PORT))
client_socket.sendall(json.dumps(message).encode('utf-8'))
print("Сообщение отправлено серверу.")
Дальше пишем простого бота на aiogram, из конфига берем информацию о кнопках меню, команды для получения данных о сервере и подключаем серверную часть агента. Получается простой бот, который принимает сообщения, пересылает их в тг, может быть настроен под нужные команды.
Дальше остается написать клиент-агента, который будет мониторить сервер и службы, и присылать оповещения.
Если будет интересно, то подготовлю статью и про клиент-агента. А еще больше информации и проектов на моем тг-канале. Вливайся в тусовку)