ربات تلگرامی برای مانیتور کردن سرورها

سیموتل نگهداری - پنل‌مدیریت
1
  • این ربات تلگرام وظیفه مانیتورینگ سرورها را برعهده دارد. با بررسی مداوم وضعیت سرورها، در صورت تغییر وضعیت، اعلان‌هایی را به مدیران و کاربران مجاز ارسال می‌کند.

قابلیت‌ها

  • مانیتورینگ مستمر سرورها از طریق پینگ و درخواست HTTP
  • ارسال اعلان به مدیران و کاربران در صورت تغییر وضعیت سرورها
  • امکان افزودن و حذف سرورها توسط مدیران بات
  • نمایش لیست سرورهای مانیتور شده
  • دریافت Chat ID کاربران

پیش‌نیازها

  • Python

کتابخانه‌های مورد نیاز: requests , sqlite3 , telebot

نصب و اجرا

  1. پکیج‌های مورد نیاز را نصب کنید:
pip install pyTelegramBotAPI requests
  1. فایل server_monitoring_bot.py را اجرا کنید:
python server_monitoring_bot.py

کد کامل بات تلگرام:

import time
import subprocess
import requests
import sqlite3
import telebot
from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton

BOT_TOKEN = "TOKEN"
ADMIN_IDS = ["ADMIN_CHAT_ID", "ADMIN_CHAT_ID2"] 
USER_IDS = ["USER_CHAT_ID", "USER_CHAT_ID2"] 
bot = telebot.TeleBot(BOT_TOKEN)


conn = sqlite3.connect("servers.db", check_same_thread=False)
c = conn.cursor()
c.execute("""
    CREATE TABLE IF NOT EXISTS servers (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        ip TEXT NOT NULL,
        last_status TEXT DEFAULT ''
    )
""")
conn.commit()

c.execute("SELECT * FROM servers WHERE ip = ?", ("192.168.1.1",))
if not c.fetchone():
    c.execute("INSERT INTO servers (name, ip, last_status) VALUES (?, ?, ?)", ("default server", "192.168.1.1", ""))
    conn.commit()

def check_server(ip):
    try:
        ping_response = subprocess.run(["ping", "-c", "3", ip], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        is_ping_ok = ping_response.returncode == 0
        
        is_http_ok = False
        try:
            response = requests.get(f"http://{ip}", timeout=3)
            is_http_ok = response.status_code == 200
        except:
            pass
        
        return is_ping_ok or is_http_ok
    except Exception as e:
        return False

def send_alert(chat_ids, message):
    for chat_id in chat_ids:
        try:
            bot.send_message(chat_id, message)
        except:
            pass

def get_chat_ids():
    return ADMIN_IDS, USER_IDS

def monitor_servers():
    while True:
        c.execute("SELECT * FROM servers")
        servers = c.fetchall()
        
        admin_chats, user_chats = get_chat_ids()
        
        for server in servers:
            server_id, name, ip, last_status = server
            is_online = check_server(ip)
            new_status = "Online" if is_online else "Offline"
            
            if new_status != last_status:
                message = f"node {name} {new_status}"
                send_alert(admin_chats + user_chats, message)
                
                c.execute("UPDATE servers SET last_status = ? WHERE id = ?", (new_status, server_id))
                conn.commit()
        
        time.sleep(60)

import threading
threading.Thread(target=monitor_servers, daemon=True).start()

def get_main_menu(is_admin):
    markup = InlineKeyboardMarkup()
    if is_admin:
        markup.add(InlineKeyboardButton("➕ افزودن سرور", callback_data="add_server"))
        markup.add(InlineKeyboardButton("❌ حذف سرور", callback_data="remove_server"))
        markup.add(InlineKeyboardButton("📜 لیست سرورها", callback_data="list_servers"))
    markup.add(InlineKeyboardButton("📌 دریافت Chat ID", callback_data="get_chat_id"))
    return markup

@bot.message_handler(commands=['start'])
def start(message):
    chat_id = str(message.chat.id)
    if chat_id in ADMIN_IDS or chat_id in USER_IDS:
        bot.send_message(chat_id, "سلام! این ربات وضعیت سرورها را مانیتور می‌کند.", reply_markup=get_main_menu(chat_id in ADMIN_IDS))
    else:
        markup = InlineKeyboardMarkup()
        markup.add(InlineKeyboardButton("📌 دریافت Chat ID", callback_data="get_chat_id"))
        bot.send_message(chat_id, "⛔ دسترسی شما مجاز نیست. لطفاً جهت فعال‌سازی با مدیریت در ارتباط باشید.", reply_markup=markup)

@bot.callback_query_handler(func=lambda call: True)
def callback_handler(call):
    chat_id = str(call.message.chat.id)
    if call.data == "get_chat_id":
        bot.send_message(chat_id, f"📌 Chat ID شما: {chat_id}")
        return
    
    if chat_id not in ADMIN_IDS:
        bot.send_message(chat_id, "⛔ شما دسترسی مدیریت بات را ندارید.")
        return
    
    if call.data == "add_server":
        bot.send_message(chat_id, "لطفاً نام و آی‌پی سرور را به فرمت زیر ارسال کنید:\nنام سرور - آی‌پی")
        bot.register_next_step_handler(call.message, add_server)
    elif call.data == "remove_server":
        bot.send_message(chat_id, "لطفاً آی‌پی سرور موردنظر برای حذف را ارسال کنید.")
        bot.register_next_step_handler(call.message, remove_server)
    elif call.data == "list_servers":
        c.execute("SELECT name, ip FROM servers")
        servers = c.fetchall()
        if servers:
            msg = "📜 لیست سرورها:\n" + "\n".join([f"{s[0]} - {s[1]}" for s in servers])
        else:
            msg = "هیچ سروری ثبت نشده است."
        bot.send_message(chat_id, msg)

def add_server(message):
    chat_id = str(message.chat.id)
    if chat_id not in ADMIN_IDS:
        bot.send_message(chat_id, "⛔ شما دسترسی مدیریت بات را ندارید.")
        return
    
    try:
        name, ip = message.text.split(" - ")
        c.execute("INSERT INTO servers (name, ip, last_status) VALUES (?, ?, ?)", (name.strip(), ip.strip(), ""))
        conn.commit()
        bot.send_message(chat_id, f"✅ سرور {name} با آی‌پی {ip} اضافه شد.", reply_markup=get_main_menu(True))
    except:
        bot.send_message(chat_id, "❌ فرمت اشتباه است. لطفاً مجدداً تلاش کنید.", reply_markup=get_main_menu(True))

def remove_server(message):
    chat_id = str(message.chat.id)
    if chat_id not in ADMIN_IDS:
        bot.send_message(chat_id, "⛔ شما دسترسی مدیریت ندارید.")
        return
    
    ip = message.text.strip()
    c.execute("SELECT * FROM servers WHERE ip = ?", (ip,))
    server = c.fetchone()
    
    if server:
        c.execute("DELETE FROM servers WHERE ip = ?", (ip,))
        conn.commit()
        bot.send_message(chat_id, f"✅ سرور با آی‌پی {ip} حذف شد.", reply_markup=get_main_menu(True))
    else:
        bot.send_message(chat_id, "❌ سروری با این آی‌پی یافت نشد. لطفاً مجدداً تلاش کنید.", reply_markup=get_main_menu(True))

bot.polling(none_stop=True)

ساختار کد

تنظیمات اولیه

در ابتدای کد، توکن ربات و لیست مدیران و کاربران مجاز تعیین شده است:

BOT_TOKEN = "توکن ربات"
ADMIN_IDS = ["آی‌دی ادمین"]
USER_IDS = ["آی‌دی کاربران"]
bot = telebot.TeleBot(BOT_TOKEN)

مدیریت پایگاه داده

یک دیتابیس sqlite3 برای ذخیره لیست سرورها ایجاد شده است:

conn = sqlite3.connect("servers.db", check_same_thread=False)
c = conn.cursor()
c.execute("""
    CREATE TABLE IF NOT EXISTS servers (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        ip TEXT NOT NULL,
        last_status TEXT DEFAULT ''
    )
""")
conn.commit()

بررسی وضعیت سرور

وضعیت هر سرور از طریق پینگ و درخواست HTTP بررسی می‌شود:

def check_server(ip):
    ping_response = subprocess.run(["ping", "-c", "3", ip], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    is_ping_ok = ping_response.returncode == 0
    
    is_http_ok = False
    try:
        response = requests.get(f"http://{ip}", timeout=3)
        is_http_ok = response.status_code == 200
    except:
        pass
    
    return is_ping_ok or is_http_ok

اجرای حلقه مانیتورینگ

هر ۶۰ ثانیه یکبار وضعیت سرورها بررسی و تغییرات احتمالی اعلام می‌شود:

def monitor_servers():
    while True:
        c.execute("SELECT * FROM servers")
        servers = c.fetchall()
        
        for server in servers:
            server_id, name, ip, last_status = server
            is_online = check_server(ip)
            new_status = "Online" if is_online else "Offline"
            
            if new_status != last_status:
                send_alert(ADMIN_IDS + USER_IDS, f"سرور {name} {new_status}")
                c.execute("UPDATE servers SET last_status = ? WHERE id = ?", (new_status, server_id))
                conn.commit()
        
        time.sleep(60)

راه‌اندازی ربات

ربات به دستورات /start و سایر دستورات مدیریت سرورها واکنش نشان می‌دهد:

@bot.message_handler(commands=['start'])
def start(message):
    chat_id = str(message.chat.id)
    if chat_id in ADMIN_IDS or chat_id in USER_IDS:
        bot.send_message(chat_id, "ربات مانیتورینگ فعال است.", reply_markup=get_main_menu(chat_id in ADMIN_IDS))
    else:
        bot.send_message(chat_id, "⛔ دسترسی شما مجاز نیست.")

نتیجه‌گیری

این ربات به سادگی قابل اجرا و سفارشی‌سازی است و به مدیران کمک می‌کند تا وضعیت سرورهای خود را به صورت لحظه‌ای بررسی کرده و در صورت بروز مشکل، سریعاً اقدام کنند.

2 پسندیده