Блог AST-SoftPro

Python-боты для мониторинга: как следить за серверами и API

30.05.2026 10 мин чтения

Введение в Python-боты для мониторинга

Мониторинг серверов и API — важная часть поддержания стабильной работы IT-инфраструктуры. Регулярная проверка доступности сервисов, время отклика (latency), статус HTTP-кодов и другие метрики позволяют оперативно выявлять сбои до их влияния на пользователей.

Одним из простых и эффективных решений является создание собственного бота мониторинга на Python с интеграцией в Telegram. Такой подход подходит для небольших команд, стартапов или внутренних систем, где нет бюджета на коммерческие решения вроде Datadog или Prometheus + Alertmanager, но требуется базовая сигнализация о проблемах.

В этой статье мы рассмотрим:

основы создания Python-бота;
проверку uptime и состояния API;
отправку алертов через Telegram;
сбор статистики в виде дашборда (с использованием JSON и простых визуализаций).

Все примеры реализованы на чистом Python, без внешних фреймворков — чтобы было понятно каждому разработчику.

Как работает система мониторинга

Простая схема работы бота включает три этапа:

Проверка сервиса — выполнение HTTP-запроса к API или серверу;
Анализ результата — сравнение с ожидаемым состоянием (например, статус 200 OK);
Отправка уведомления — если состояние изменилось, бот отправляет сообщение в Telegram.

Для реализации этого достаточно базовых модулей Python: requests, time и logging. В качестве канала уведомлений используется Telegram Bot API — он бесплатен и легко интегрируется.

Настройка Telegram-бота

Чтобы начать, нужно создать бота через @BotFather в Telegram:

Напишите /newbot;
Укажите имя (например, ServerMonitor) и username (@ServerMonitorBot);
Получите токен — он понадобится для авторизации.

Пример получения токена: 1234567890:AABcD_eFgHiJkLmNoPqRsTuvWxYzAbCdEfG. Сохраните его в переменной окружения или файле .env.

Подключение к Telegram API

Для отправки сообщений используется requests:

telegram_token = "1234567890:AABcD_eFgHiJkLmNoPqRsTuvWxYzAbCdEfG"
telega_url = f"https://api.telegram.org/bot{telegram_token}/sendMessage"

def send_telegram(message, chat_id=None):
    payload = {
        "chat_id": chat_id,
        "text": message
    }
    requests.post(telega_url, data=payload)

Проверка uptime и состояния API

Основная задача бота — регулярно проверять доступность сервисов. Для этого используется HTTP-запрос к целевому URL с замером времени отклика:

import requests
import time

def check_service(url, timeout=5):
    """Проверяет доступность сервиса и возвращает статус."""
    start = time.time()
    try:
        response = requests.get(url, timeout=timeout)
        latency = round((time.time() - start) * 1000, 2)
        return {
            "url": url,
            "status": response.status_code,
            "latency": latency,
            "ok": response.status_code == 200
        }
    except requests.RequestException as e:
        return {
            "url": url,
            "status": "error",
            "latency": None,
            "ok": False,
            "error": str(e)
        }

Функция возвращает словарь с результатом: статус-код, время отклика (в миллисекундах) и флаг доступности. Если сервис не отвечает или превышает таймаут — возвращается ошибка.

Отправка алертов

Теперь объединим проверку и отправку уведомлений. Бот должен отправлять алерт только при изменении состояния — например, когда сервис перешёл из доступного в недоступный:

def monitor_service(url, chat_id, threshold_ms=3000):
    """Мониторит сервис и отправляет алерт при проблемах."""
    result = check_service(url)

    if not result["ok"]:
        message = (
            f"🔴 *ALERT*: {result['url']} недоступен!\n"
            f"Статус: {result['status']}\n"
            f"{'Ошибка: ' + result.get('error', '') if 'error' in result else ''}"
        )
        send_telegram(message, chat_id)
        return result

    if result["latency"] and result["latency"] > threshold_ms:
        message = (
            f"🟡 *WARNING*: {result['url']}\n"
            f"Высокая задержка: {result['latency']} мс"
        )
        send_telegram(message, chat_id)

    return result

Параметр threshold_ms задаёт порог задержки (по умолчанию 3000 мс). Если сервис отвечает, но медленнее порога — отправляется предупреждение. Если сервис недоступен — отправляется критический алерт.

Основной цикл мониторинга

Для автоматической проверки нескольких сервисов создаём цикл с интервалом:

import json
import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s — %(levelname)s — %(message)s"
)

SERVICES = [
    "https://api.example.com/health",
    "https://myapp.example.com/status",
    "https://db.example.com/ping"
]

CHAT_ID = "123456789"
CHECK_INTERVAL = 60  # секунд

def main():
    """Основной цикл мониторинга."""
    logging.info("Мониторинг запущен")

    while True:
        for url in SERVICES:
            try:
                result = monitor_service(url, CHAT_ID)
                logging.info(f"{url} — статус: {result['status']}, "
                           f"задержка: {result['latency']} мс")
            except Exception as e:
                logging.error(f"Ошибка проверки {url}: {e}")

        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    main()

Скрипт проверяет все сервисы из списка SERVICES с интервалом CHECK_INTERVAL секунд. Результаты логируются, а алерты отправляются в Telegram при обнаружении проблем.

Сбор статистики и простой дашборд

Для наглядности можно сохранять результаты проверок в JSON-файл и строить простой отчёт:

from datetime import datetime

STATS_FILE = "monitoring_stats.json"

def save_stats(result):
    """Сохраняет результат проверки в JSON-файл."""
    stats = []
    if os.path.exists(STATS_FILE):
        with open(STATS_FILE, "r") as f:
            stats = json.load(f)

    stats.append({
        "timestamp": datetime.now().isoformat(),
        "url": result["url"],
        "status": result["status"],
        "latency": result["latency"],
        "ok": result["ok"]
    })

    # Храним последние 100 записей
    stats = stats[-100:]

    with open(STATS_FILE, "w") as f:
        json.dump(stats, f, indent=2)

def get_dashboard():
    """Формирует простой текстовый дашборд."""
    if not os.path.exists(STATS_FILE):
        return "Нет данных"

    with open(STATS_FILE, "r") as f:
        stats = json.load(f)

    lines = ["📊 *Дашборд мониторинга*\n"]
    for url in SERVICES:
        url_stats = [s for s in stats if s["url"] == url]
        if not url_stats:
            continue

        ok_count = sum(1 for s in url_stats if s["ok"])
        total = len(url_stats)
        avg_latency = sum(s["latency"] or 0 for s in url_stats) / total

        lines.append(
            f"🟢 {url}\n"
            f"  Uptime: {ok_count}/{total} ({ok_count/total*100:.0f}%)\n"
            f"  Avg latency: {avg_latency:.0f} мс"
        )

    return "\n".join(lines)

Добавим вызов save_stats(result) в функцию monitor_service, а отправку дашборда — по расписанию (например, раз в час):

import schedule

def send_dashboard():
    """Отправляет дашборд в Telegram."""
    dashboard = get_dashboard()
    send_telegram(dashboard, CHAT_ID)

# Запускать дашборд каждый час
schedule.every(60).minutes.do(send_dashboard)

Заключение

Python-бот для мониторинга — простое и эффективное решение для небольших команд и внутренних систем. Основные преимущества:

Низкая стоимость — не требует коммерческих сервисов
Гибкость — легко настроить под любые сервисы
Простота — чистый Python без внешних зависимостей
Мгновенные уведомления — алерты в Telegram в реальном времени

Для масштабирования можно добавить:

Базу данных для хранения истории (SQLite или PostgreSQL)
Веб-интерфейс для визуализации (Flask + Chart.js)
Интеграцию с Prometheus или Grafana
Параллельные проверки через asyncio и aiohttp

Начните с простого бота, а затем расширяйте функционал по мере необходимости.

Другие записи блога

Все статьи

К списку статей