Xây dựng Network Monitor Tool với Python: Giám sát Hệ thống Mạng Chuyên Nghiệp

Xin chào các bạn! Hôm nay tôi sẽ hướng dẫn các bạn xây dựng một công cụ giám sát mạng chuyên nghiệp bằng Python. Tool này không chỉ giúp theo dõi trạng thái kết nối của nhiều thiết bị mà còn tích hợp tính năng cảnh báo qua Telegram và ghi log chi tiết.

Tổng quan về Network Monitor Tool

{tocify} 

1. Giới thiệu

Tool Ping kiểm tra hệ thống mạng là một công cụ giúp:

• Giám sát tình trạng kết nối của nhiều máy chủ/thiết bị
• Gửi cảnh báo qua Telegram khi phát hiện sự cố
• Lưu log để theo dõi và phân tích
• Tự động khởi động lại khi gặp sự cố

Các tính năng chính:

• Giám sát đồng thời nhiều thiết bị mạng (IPv4/IPv6)
• Gửi cảnh báo real-time qua Telegram
• Hệ thống logging hoàn chỉnh
• Tự động khôi phục khi gặp sự cố
• Giao diện CLI thân thiện
• Hỗ trợ chạy như Windows Service

2. Chuẩn bị môi trường

2.1. Yêu cầu hệ thống

• Python 3.7 trở lên
• Windows/Linux/MacOS

2.2. Cài đặt môi trường

# Tạo thư mục project
mkdir ping_monitor
cd ping_monitor

# Tạo môi trường ảo
python -m venv venv

# Kích hoạt môi trường ảo
# Windows:
.\venv\Scripts\activate
# Linux/Mac:
source venv/bin/activate

# Cài đặt thư viện
pip install requests
pip install pyinstaller
pip install python-telegram-bot
pip install colorama

# Lưu requirements
pip freeze > requirements.txt

3. Cấu trúc project

ping_monitor/
├── src/
│   ├── __init__.py
│   ├── ping_monitor.py
│   ├── telegram_handler.py
│   └── logger.py
├── config/
│   ├── config.json
│   └── hosts.json
├── assets/
│   └── icon.ico # (tùy chọn) Icon cho file exe
├── venv/
├── build.py
├── requirements.txt
└── README.md

3.1. Tạo cấu trúc thư mục

mkdir src config logs assets
type nul > src\__init__.py
type nul > src\ping_monitor.py
type nul > src\telegram_handler.py
type nul > src\logger.py
type nul > config\config.json
type nul > config\hosts.json
type nul > build.py
type nul > README.md

3.2. Tạo file cấu hình

config/config.json:

{
    "telegram": {
        "bot_token": "YOUR_BOT_TOKEN",
        "chat_id": "YOUR_CHAT_ID"
    },
    "ping": {
        "interval": 60,
        "timeout": 5,
        "retry": 3
    },
    "log": {
        "file": "logs/ping.log",
        "level": "INFO"
    }
}

config/hosts.json:

{
    "hosts": [
        {
            "name": "Router",
            "ip": "192.168.1.1",
            "type": "LOCAL",
            "description": "Default Gateway Router"
        },
        {
            "name": "Server",
            "ip": "192.168.1.100",
            "type": "LOCAL",
            "description": "Local Server"
        },
        {
            "name": "VN_Zing",
            "ip": "zing.vn",
            "type": "DOMAIN",
            "description": "Zing Vietnam Website"
        },
        {
            "name": "QT_Google_DNS",
            "ip": "8.8.8.8",
            "type": "PUBLIC",
            "description": "Google Public DNS"
        },
        {
            "name": "QT_Cloudflare_DNS",
            "ip": "1.1.1.1",
            "type": "PUBLIC",
            "description": "Cloudflare Public DNS"
        },
        {
            "name": "QT_Facebook",
            "ip": "facebook.com",
            "type": "DOMAIN",
            "description": "Facebook Website"
        }
    ],
    "groups": {
        "LOCAL": "Thiết bị mạng nội bộ",
        "PUBLIC": "Dịch vụ công cộng quốc tế",
        "DOMAIN": "Tên miền website"
    }
}

4. Code chi tiết

4.1. Logger (src/logger.py)

import logging
import os
from datetime import datetime

class Logger:
    def __init__(self, log_file):
        self.log_file = log_file
        self._setup_logger()

    def _setup_logger(self):
        # Tạo thư mục logs nếu chưa tồn tại
        os.makedirs(os.path.dirname(self.log_file), exist_ok=True)

        # Cấu hình logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(self.log_file),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)

    def info(self, message):
        self.logger.info(message)

    def error(self, message):
        self.logger.error(message)

    def warning(self, message):
        self.logger.warning(message)

    def get_logs(self, lines=100):
        if not os.path.exists(self.log_file):
            return []
        with open(self.log_file, 'r') as f:
            return f.readlines()[-lines:]

4.2. Telegram Handler (src/telegram_handler.py)

import requests
from datetime import datetime

class TelegramHandler:
    def __init__(self, bot_token, chat_id):
        self.bot_token = bot_token
        self.chat_id = chat_id
        self.api_url = f"https://api.telegram.org/bot{bot_token}"

    def send_message(self, message):
        try:
            url = f"{self.api_url}/sendMessage"
            data = {
                "chat_id": self.chat_id,
                "text": message,
                "parse_mode": "HTML"
            }
            response = requests.post(url, json=data)
            response.raise_for_status()
            return True
        except Exception as e:
            print(f"Error sending Telegram message: {e}")
            return False

    def send_alert(self, host, status, previous_status):
        if status == "DOWN":
            emoji = "🔴"
            alert_type = "CẢNH BÁO"
            status_message = "Mất kết nối"
        else:  # UP from DOWN
            emoji = "🟢"
            alert_type = "PHỤC HỒI"
            status_message = "Đã phục hồi kết nối"

        message = (
            f"{emoji} <b>{alert_type}</b>\n"
            f"Tên: {host['name']}\n"
            f"IP/Domain: {host['ip']}\n"
            f"Loại: {host['type']}\n"
            f"Mô tả: {host['description']}\n"
            f"Trạng thái: {status_message}\n"
            f"Thời gian: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        )
        return self.send_message(message)

4.3. Main Program (src/ping_monitor.py)

import os
import json
import time
import platform
import subprocess
from datetime import datetime
from logger import Logger
from telegram_handler import TelegramHandler

class PingMonitor:
    def __init__(self):
        self.load_config()
        self.logger = Logger(self.config['log']['file'])
        self.telegram = TelegramHandler(
            self.config['telegram']['bot_token'],
            self.config['telegram']['chat_id']
        )
        self.hosts_status = {}

    def load_config(self):
        with open('config/config.json', 'r') as f:
            self.config = json.load(f)
        with open('config/hosts.json', 'r') as f:
            self.hosts = json.load(f)['hosts']

    def ping(self, host):
        ip = host['ip']
        ping_count = '-n' if platform.system().lower() == 'windows' else '-c'
        
        try:
            cmd = ['ping', ping_count, '1', '-w', '1000', ip]
            result = subprocess.run(cmd, 
                                 stdout=subprocess.PIPE, 
                                 stderr=subprocess.PIPE)
            return result.returncode == 0
        except Exception as e:
            self.logger.error(f"Error pinging {ip}: {str(e)}")
            return False

 def check_host(self, host):
        current_status = "UP" if self.ping(host) else "DOWN"
        previous_status = self.hosts_status.get(host['ip'], "UNKNOWN")

        if current_status != previous_status:
            self.hosts_status[host['ip']] = current_status
            self.logger.info(
                f"Status change - Host: {host['name']} ({host['ip']}) "
                f"Type: {host['type']} Status: {current_status}"
            )
            
            # Chỉ gửi thông báo trong 2 trường hợp:
            # 1. Khi trạng thái chuyển sang DOWN
            # 2. Khi trạng thái từ DOWN chuyển sang UP
            if current_status == "DOWN" or (current_status == "UP" and previous_status == "DOWN"):
                self.telegram.send_alert(host, current_status, previous_status)

        return current_status

    def monitor(self):
        self.logger.info("Starting ping monitor...")
        
        try:
            while True:
                for host in self.hosts:
                    status = self.check_host(host)
                    print(f"{datetime.now().strftime('%H:%M:%S')} - "
                          f"{host['name']} ({host['ip']}): {status}")
                
                time.sleep(self.config['ping']['interval'])
        
        except KeyboardInterrupt:
            self.logger.info("Stopping ping monitor...")
        except Exception as e:
            self.logger.error(f"Error in monitor loop: {str(e)}")
            raise

def main():
    monitor = PingMonitor()
    monitor.monitor()

if __name__ == "__main__":
    main()

4.4. Build Script (build.py)

import os
import shutil
import PyInstaller.__main__
from datetime import datetime

def clean_dist():
    """Xóa thư mục dist và build cũ nếu tồn tại"""
    dirs_to_clean = ['dist', 'build']
    for dir_name in dirs_to_clean:
        if os.path.exists(dir_name):
            shutil.rmtree(dir_name)
            print(f"Đã xóa thư mục {dir_name}")
    
    # Xóa file spec nếu tồn tại
    spec_file = "PingMonitor.spec"
    if os.path.exists(spec_file):
        os.remove(spec_file)
        print(f"Đã xóa file {spec_file}")

def create_version_file():
    """Tạo file version.txt"""
    version = datetime.now().strftime("%Y.%m.%d")
    version_file = os.path.join(os.getcwd(), "version.txt")
    with open(version_file, "w") as f:
        f.write(version)
    return version

def copy_configs():
    """Copy thư mục config vào dist"""
    if not os.path.exists("dist/config"):
        os.makedirs("dist/config")
    
    config_files = [
        "config/config.json",
        "config/hosts.json"
    ]
    
    for file in config_files:
        if os.path.exists(file):
            shutil.copy2(file, "dist/config/")
            print(f"Đã copy {file} vào dist/config/")
        else:
            print(f"Không tìm thấy file {file}")

def build_exe():
    """Build file exe using PyInstaller"""
    version = create_version_file()
    
    # Đường dẫn tới các file nguồn
    main_file = os.path.join(os.getcwd(), 'src', 'ping_monitor.py')
    version_file = os.path.join(os.getcwd(), 'version.txt')
    
    PyInstaller.__main__.run([
        main_file,                            # File chính
        '--name=PingMonitor',                 # Tên file exe
        '--onefile',                          # Đóng gói thành 1 file
        '--console',                          # Hiển thị console để debug
        '--clean',                            # Xóa cache trước khi build
        '--distpath=dist',                    # Thư mục output
        '--workpath=build',                   # Thư mục build tạm
        f'--add-data={version_file};.',       # Thêm file version
        '--add-data=src/logger.py;.',         # Thêm các module phụ thuộc
        '--add-data=src/telegram_handler.py;.',
        '--hidden-import=requests',           # Thêm các module ẩn
        '--hidden-import=urllib3',
        '--hidden-import=chardet',
        '--hidden-import=certifi',
        '--hidden-import=idna',
        '--hidden-import=charset_normalizer'
    ])

def main():
    try:
        print("Bắt đầu quá trình build...")
        
        # Đảm bảo đang ở thư mục gốc của project
        project_root = os.path.dirname(os.path.abspath(__file__))
        os.chdir(project_root)
        
        # Xóa các thư mục cũ
        clean_dist()
        
        # Build file exe
        build_exe()
        
        # Copy config files
        copy_configs()
        
        print("\nBuild thành công!")
        print("File exe được tạo tại: dist/PingMonitor.exe")
        
    except Exception as e:
        print(f"\nLỗi trong quá trình build: {str(e)}")
        return 1
    
    return 0

if __name__ == "__main__":
    exit(main())

5. Build và triển khai

5.1. Build thành file exe

# Kích hoạt môi trường ảo
.\venv\Scripts\activate

# Build
python build.py

5.2. Chuẩn bị triển khai

1. Copy các file/thư mục sau vào thư mục triển khai:

• dist/ping_monitor.exe
• config/
• logs/

2. Chỉnh sửa config/config.json với thông tin thực tế:

• Bot token Telegram
• Chat ID
• Các thông số khác

3. Cập nhật config/hosts.json với danh sách host cần giám sát

6. Hướng dẫn sử dụng

6.1. Chạy chương trình

# Chạy trực tiếp từ source
python src/ping_monitor.py

# Hoặc chạy file exe
ping_monitor.exe

6.2. Cấu hình Windows Service

1. Tạo file ping_monitor.bat:

@echo off
cd /d %~dp0
ping_monitor.exe

2. Tạo Windows Service sử dụng Task Scheduler:

• Open Task Scheduler
• Create Basic Task
• Trigger: At startup
• Action: Start a program -> chọn file .bat
• Finish

7. Xử lý lỗi thường gặp

7.1. Lỗi Module Not Found

pip install -r requirements.txt

7.2. Lỗi Permission Denied

# Chạy với quyền Administrator
# Hoặc kiểm tra quyền truy cập thư mục logs

7.3. Lỗi Telegram

• Kiểm tra bot_token và chat_id
• Kiểm tra kết nối internet
• Kiểm tra logs để xem chi tiết lỗi

7.4. Lỗi Ping

• Kiểm tra quyền ICMP
• Kiểm tra firewall
• Đảm bảo định dạng IP đúng

Tổng kết

Tool Ping Monitor là một giải pháp đơn giản nhưng hiệu quả để giám sát hệ thống mạng. Với khả năng:

• Giám sát nhiều host
• Gửi cảnh báo qua Telegram
• Lưu log chi tiết
• Dễ dàng cấu hình và triển khai

Tool có thể được tùy chỉnh và mở rộng thêm các tính năng như:

• Giao diện web
• Thống kê và báo cáo
• Hỗ trợ các phương thức cảnh báo khác
• Tích hợp với các hệ thống monitoring khác

Mọi đóng góp và phản hồi đều được hoan nghênh để cải thiện tool ngày càng tốt hơn.