diff --git a/monitor.py b/monitor.py index b2f82d4..f5d6852 100644 --- a/monitor.py +++ b/monitor.py @@ -1,21 +1,31 @@ #!/usr/bin/env python3 """ MikroTik IP Monitor — читает адреса из /ip firewall address-list, -пингует их через MikroTik API и шлёт уведомления в Telegram. +пингует параллельно, шлёт уведомления в Telegram и MAX (VK Teams). + +Возможности: + - Параллельный пинг всех хостов + - Повторные уведомления о даунтайме + - Время даунтайма при восстановлении + - Ежедневный отчёт + - Команды /status, /help через Telegram и MAX + - Веб-дашборд на порту 8080 """ import os import time import logging import asyncio -from datetime import datetime +import threading +from datetime import datetime, timedelta from typing import Optional +from http.server import HTTPServer, BaseHTTPRequestHandler import routeros_api import httpx # ────────────────────────────────────────────── -# Настройки из переменных окружения +# Настройки # ────────────────────────────────────────────── MIKROTIK_HOST = os.environ["MIKROTIK_HOST"] MIKROTIK_PORT = int(os.getenv("MIKROTIK_PORT", "8728")) @@ -26,18 +36,18 @@ TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "") TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "") TELEGRAM_API_URL = os.getenv("TELEGRAM_API_URL", "https://api.telegram.org") -# MAX (VK Teams) — опционально, если не заданы — уведомления не отправляются MAX_BOT_TOKEN = os.getenv("MAX_BOT_TOKEN", "") MAX_CHAT_ID = os.getenv("MAX_CHAT_ID", "") MAX_API_URL = os.getenv("MAX_API_URL", "https://platform-api.max.ru") -# Имя address-list в MikroTik (/ip firewall address-list) ADDRESS_LIST_NAME = os.environ["ADDRESS_LIST_NAME"] - CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "60")) PING_COUNT = int(os.getenv("PING_COUNT", "3")) PING_TIMEOUT_MS = int(os.getenv("PING_TIMEOUT_MS", "1000")) FAILURE_THRESHOLD = int(os.getenv("FAILURE_THRESHOLD", "2")) +REPEAT_NOTIFY_MIN = int(os.getenv("REPEAT_NOTIFY_MIN", "30")) +DAILY_REPORT_TIME = os.getenv("DAILY_REPORT_TIME", "09:00") +WEB_PORT = int(os.getenv("WEB_PORT", "8080")) # ────────────────────────────────────────────── # Логирование @@ -49,9 +59,20 @@ logging.basicConfig( ) log = logging.getLogger(__name__) +# ────────────────────────────────────────────── +# Глобальное состояние +# ────────────────────────────────────────────── +hosts_state: dict[str, dict] = {} +router_identity: str = MIKROTIK_HOST +last_check_time: Optional[datetime] = None +daily_report_sent_date: Optional[str] = None +_tg_bot_username: str = "" +tg_last_update_id: int = 0 +max_last_marker: int = 0 + # ────────────────────────────────────────────── -# Telegram +# Уведомления # ────────────────────────────────────────────── async def send_telegram(message: str) -> None: if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: @@ -61,57 +82,41 @@ async def send_telegram(message: str) -> None: async with httpx.AsyncClient(timeout=10) as client: try: resp = await client.post(url, data=payload) - log.info("Telegram response %s: %s", resp.status_code, resp.text) resp.raise_for_status() - log.info("Telegram: сообщение отправлено") + log.info("Telegram: отправлено") except httpx.HTTPStatusError as exc: - log.error("Telegram: ошибка %s — %s", exc.response.status_code, exc.response.text) + log.error("Telegram: %s — %s", exc.response.status_code, exc.response.text) except Exception as exc: - log.error("Telegram: ошибка отправки — %s", exc) + log.error("Telegram: %s", exc) async def send_max(message: str) -> None: - """Отправляет уведомление в MAX (VK Teams).""" if not MAX_BOT_TOKEN or not MAX_CHAT_ID: return - import re - # chat_id и user_id — query-параметры, не тело запроса - params = {"chat_id": MAX_CHAT_ID} - headers = { - "Authorization": MAX_BOT_TOKEN, - "Content-Type": "application/json", - } - # MAX поддерживает HTML — оставляем форматирование + headers = {"Authorization": MAX_BOT_TOKEN, "Content-Type": "application/json"} payload = {"text": message, "format": "html"} async with httpx.AsyncClient(timeout=10) as client: try: resp = await client.post( f"{MAX_API_URL}/messages", - headers=headers, - params=params, - json=payload, + headers=headers, params={"chat_id": MAX_CHAT_ID}, json=payload, ) - log.info("MAX response %s: %s", resp.status_code, resp.text) resp.raise_for_status() - log.info("MAX: сообщение отправлено") + log.info("MAX: отправлено") except Exception as exc: - log.error("MAX: ошибка отправки — %s", exc) + log.error("MAX: %s", exc) def notify(message: str) -> None: - async def _send_all(): - await asyncio.gather( - send_telegram(message), - send_max(message), - ) - asyncio.run(_send_all()) + async def _all(): + await asyncio.gather(send_telegram(message), send_max(message)) + asyncio.run(_all()) # ────────────────────────────────────────────── -# MikroTik: получить список IP из address-list +# MikroTik утилиты # ────────────────────────────────────────────── def decode(val) -> str: - """Декодирует bytes или str. Пробует UTF-8, затем Windows-1251.""" if isinstance(val, bytes): try: return val.decode("utf-8") @@ -121,18 +126,9 @@ def decode(val) -> str: def fetch_address_list(api) -> list[tuple[str, str]]: - """ - Читает /ip firewall address-list и возвращает список (ip, comment) - из записей с именем ADDRESS_LIST_NAME (пропускает отключённые). - """ - resource = api.get_binary_resource("/ip/firewall/address-list") - all_entries = resource.get() - ip_list = [] - for entry in all_entries: - # Ключи строковые, значения — bytes - entry_list = decode(entry.get("list", b"")) - if entry_list != ADDRESS_LIST_NAME: + for entry in api.get_binary_resource("/ip/firewall/address-list").get(): + if decode(entry.get("list", b"")) != ADDRESS_LIST_NAME: continue if decode(entry.get("disabled", b"false")) == "true": continue @@ -142,105 +138,297 @@ def fetch_address_list(api) -> list[tuple[str, str]]: comment = decode(entry.get("comment", b"")).strip() if address: ip_list.append((address, comment)) - return ip_list -# ────────────────────────────────────────────── -# MikroTik: пинг через API -# ────────────────────────────────────────────── -def mikrotik_ping(api, address: str) -> Optional[float]: - """ - Возвращает avg-rtt (мс) или None если хост недоступен. - RouterOS API: /ping — это команда, вызывается через get_binary_resource. - """ - try: - binary_resource = api.get_binary_resource("/") - results = binary_resource.call( - "ping", - { - "address": address.encode(), - "count": str(PING_COUNT).encode(), - "interval": b"0.2", - }, - ) - - def parse_rtt(val) -> Optional[float]: - """Парсит '3ms674us', '2ms', '500us' → миллисекунды (float).""" - if isinstance(val, bytes): - val = val.decode() - val = val.strip() - if not val: - return None - ms, us = 0.0, 0.0 - if "ms" in val: - parts = val.split("ms") - try: - ms = float(parts[0]) - except ValueError: - pass - if len(parts) > 1 and "us" in parts[1]: - try: - us = float(parts[1].replace("us", "")) - except ValueError: - pass - elif "us" in val: - try: - us = float(val.replace("us", "")) - except ValueError: - pass - total = ms + us / 1000.0 - return total if total > 0 else None - - rtts = [] - for row in results: - # Берём avg-rtt из последней строки (она накопительная) - rtt = parse_rtt(row.get("avg-rtt", "")) - if rtt is not None: - rtts.append(rtt) - - if rtts: - return rtts[-1] # последняя строка содержит финальный avg - return None - - except Exception as exc: - log.warning("Ошибка пинга %s: %s", address, exc) - return None - - -# ────────────────────────────────────────────── -# MikroTik: получить system identity -# ────────────────────────────────────────────── def fetch_identity(api) -> str: try: - resource = api.get_binary_resource("/system/identity") - result = resource.get() - if result: - return decode(result[0].get("name", MIKROTIK_HOST.encode())) - return MIKROTIK_HOST + result = api.get_binary_resource("/system/identity").get() + return decode(result[0].get("name", MIKROTIK_HOST.encode())) if result else MIKROTIK_HOST except Exception: return MIKROTIK_HOST +def parse_rtt(val) -> Optional[float]: + if isinstance(val, bytes): + val = val.decode() + val = val.strip() + if not val: + return None + ms, us = 0.0, 0.0 + if "ms" in val: + parts = val.split("ms") + try: + ms = float(parts[0]) + except ValueError: + pass + if len(parts) > 1 and "us" in parts[1]: + try: + us = float(parts[1].replace("us", "")) + except ValueError: + pass + elif "us" in val: + try: + us = float(val.replace("us", "")) + except ValueError: + pass + total = ms + us / 1000.0 + return total if total > 0 else None + + +def mikrotik_ping(api, address: str) -> Optional[float]: + try: + results = api.get_binary_resource("/").call( + "ping", + {"address": address.encode(), "count": str(PING_COUNT).encode(), "interval": b"0.2"}, + ) + rtts = [r for r in (parse_rtt(row.get("avg-rtt", "")) for row in results) if r is not None] + return rtts[-1] if rtts else None + except Exception as exc: + log.warning("Пинг %s: %s", address, exc) + return None + + +# ────────────────────────────────────────────── +# Параллельный пинг +# ────────────────────────────────────────────── +async def ping_all_async(api, ip_list: list[tuple[str, str]]) -> dict[str, Optional[float]]: + loop = asyncio.get_event_loop() + tasks = {ip: loop.run_in_executor(None, mikrotik_ping, api, ip) for ip, _ in ip_list} + return {ip: await task for ip, task in tasks.items()} + + +# ────────────────────────────────────────────── +# Форматирование +# ────────────────────────────────────────────── +def fmt_duration(td: timedelta) -> str: + total = int(td.total_seconds()) + if total < 60: + return f"{total} сек" + elif total < 3600: + return f"{total // 60} мин" + else: + h, m = divmod(total // 60, 60) + return f"{h} ч {m} мин" if m else f"{h} ч" + + +def fmt_label(ip: str) -> str: + comment = hosts_state.get(ip, {}).get("comment", "") + return f"{ip} ({comment})" if comment else ip + + +# ────────────────────────────────────────────── +# Статус-отчёт +# ────────────────────────────────────────────── +def build_status_report(title: str) -> str: + now = datetime.now() + lines = [f"{title}", f"🔧 Роутер: {router_identity}", ""] + down = [ip for ip, s in hosts_state.items() if s["status"] == "down"] + up = [ip for ip, s in hosts_state.items() if s["status"] == "up"] + + if down: + lines.append(f"🔴 Недоступны ({len(down)}):") + for ip in down: + ds = hosts_state[ip].get("down_since") + dur = fmt_duration(now - ds) if ds else "?" + lines.append(f" • {fmt_label(ip)} — {dur}") + lines.append("") + if up: + lines.append(f"✅ Доступны ({len(up)}):") + for ip in up: + rtt = hosts_state[ip].get("rtt") + lines.append(f" • {fmt_label(ip)}" + (f" {rtt:.1f} ms" if rtt else "")) + + lines += ["", f"⏱ {now.strftime('%Y-%m-%d %H:%M:%S')}"] + return "\n".join(lines) + + +# ────────────────────────────────────────────── +# Команды ботов +# ────────────────────────────────────────────── +async def process_telegram_commands() -> None: + global tg_last_update_id, _tg_bot_username + if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: + return + async with httpx.AsyncClient(timeout=15) as client: + try: + resp = await client.get( + f"{TELEGRAM_API_URL}/bot{TELEGRAM_BOT_TOKEN}/getUpdates", + params={"timeout": 1, "offset": tg_last_update_id + 1}, + ) + data = resp.json() + for update in data.get("result", []): + tg_last_update_id = update["update_id"] + text = update.get("message", {}).get("text", "").strip().lower().split("@")[0] + if text == "/status": + await send_telegram(build_status_report("📊 Текущий статус")) + elif text == "/help": + await send_telegram( + "🤖 Доступные команды:\n" + "/status — статус всех хостов\n" + "/help — это сообщение" + ) + except Exception as exc: + log.debug("TG polling: %s", exc) + + +async def process_max_commands() -> None: + global max_last_marker + if not MAX_BOT_TOKEN or not MAX_CHAT_ID: + return + params = {"access_token": MAX_BOT_TOKEN} + if max_last_marker: + params["marker"] = max_last_marker + async with httpx.AsyncClient(timeout=15) as client: + try: + resp = await client.get(f"{MAX_API_URL}/updates", params=params) + data = resp.json() + max_last_marker = data.get("marker", max_last_marker) + for update in data.get("updates", []): + if update.get("update_type") != "message_created": + continue + text = update.get("message", {}).get("body", {}).get("text", "").strip().lower() + if text == "/status": + await send_max(build_status_report("📊 Текущий статус")) + elif text == "/help": + await send_max( + "🤖 Доступные команды:\n" + "/status — статус всех хостов\n" + "/help — это сообщение" + ) + except Exception as exc: + log.debug("MAX polling: %s", exc) + + +async def process_bot_commands() -> None: + await asyncio.gather(process_telegram_commands(), process_max_commands()) + + +# ────────────────────────────────────────────── +# Ежедневный отчёт +# ────────────────────────────────────────────── +def check_daily_report() -> None: + global daily_report_sent_date + now = datetime.now() + today = now.strftime("%Y-%m-%d") + if now.strftime("%H:%M") == DAILY_REPORT_TIME and daily_report_sent_date != today: + daily_report_sent_date = today + notify(build_status_report(f"📅 Ежедневный отчёт — {today}")) + log.info("Ежедневный отчёт отправлен") + + +# ────────────────────────────────────────────── +# Веб-дашборд +# ────────────────────────────────────────────── +class DashboardHandler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_GET(self): + if self.path not in ("/", "/status"): + self.send_response(404) + self.end_headers() + return + + now = datetime.now() + rows = "" + for ip, state in hosts_state.items(): + comment = state.get("comment", "") + status = state.get("status", "unknown") + rtt = state.get("rtt") + ds = state.get("down_since") + ls = state.get("last_seen") + + if status == "up": + badge = '● UP' + rtt_cell = f"{rtt:.1f} ms" if rtt else "—" + dur_cell = "—" + elif status == "down": + badge = '● DOWN' + rtt_cell = "—" + dur_cell = fmt_duration(now - ds) if ds else "—" + else: + badge = '● ?' + rtt_cell = dur_cell = "—" + + rows += f""" + {ip}{comment}{badge} + {rtt_cell}{dur_cell} + {ls.strftime('%H:%M:%S') if ls else '—'}""" + + total = len(hosts_state) + up_count = sum(1 for s in hosts_state.values() if s.get("status") == "up") + down_count= sum(1 for s in hosts_state.values() if s.get("status") == "down") + last_str = last_check_time.strftime("%Y-%m-%d %H:%M:%S") if last_check_time else "—" + + html = f""" + + +MikroTik Monitor — {router_identity} + +

🔍 MikroTik Monitor

+
Роутер: {router_identity} ({MIKROTIK_HOST})  |  +List: {ADDRESS_LIST_NAME}  |  Проверка: {last_str}
+
+
{total}
Всего
+
{up_count}
Доступны
+
{down_count}
Недоступны
+
+ + +{rows}
IPИмяСтатусRTTДаунтаймПоследний ответ
+
Автообновление каждые {CHECK_INTERVAL} сек
+""" + + body = html.encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +def start_web_server() -> None: + server = HTTPServer(("0.0.0.0", WEB_PORT), DashboardHandler) + log.info("Веб-дашборд: http://0.0.0.0:%d", WEB_PORT) + server.serve_forever() + + # ────────────────────────────────────────────── # Основной цикл # ────────────────────────────────────────────── def main() -> None: - log.info("=== MikroTik Monitor запущен ===") - log.info("Address-list: '%s'", ADDRESS_LIST_NAME) - log.info("Интервал: %d сек", CHECK_INTERVAL) + global router_identity, last_check_time - # Состояние хостов: счётчик ошибок и флаг DOWN - # Обновляется динамически при каждом чтении address-list - failure_count: dict[str, int] = {} - host_down: dict[str, bool] = {} + log.info("=== MikroTik Monitor запущен ===") + log.info("Address-list: '%s' Интервал: %d сек", ADDRESS_LIST_NAME, CHECK_INTERVAL) + + threading.Thread(target=start_web_server, daemon=True).start() startup_notified = False while True: pool = None try: - log.info("Подключение к MikroTik %s:%d …", MIKROTIK_HOST, MIKROTIK_PORT) pool = routeros_api.RouterOsApiPool( MIKROTIK_HOST, username=MIKROTIK_USER, @@ -251,77 +439,104 @@ def main() -> None: api = pool.get_api() log.info("Подключено к MikroTik") - # Читаем актуальный список адресов с MikroTik ip_list = fetch_address_list(api) - # Стартовое сообщение — отправляем только после успешного подключения if not startup_notified: - identity = fetch_identity(api) + router_identity = fetch_identity(api) ip_lines = "\n".join( f" • {ip} ({comment})" if comment else f" • {ip}" for ip, comment in ip_list ) or " (список пуст)" notify( - f"🟢 MikroTik Monitor запущен\n" - f"\n" - f"🔧 Роутер: {identity} ({MIKROTIK_HOST})\n" - f"📋 Address-list: {ADDRESS_LIST_NAME}\n" - f"\n" - f"Мониторинг адресов:\n{ip_lines}\n" - f"\n" + f"🟢 MikroTik Monitor запущен\n\n" + f"🔧 Роутер: {router_identity} ({MIKROTIK_HOST})\n" + f"📋 Address-list: {ADDRESS_LIST_NAME}\n\n" + f"Мониторинг адресов:\n{ip_lines}\n\n" f"⏱ Интервал: {CHECK_INTERVAL} сек" ) startup_notified = True if not ip_list: - log.warning("Address-list '%s' пуст или не найден!", ADDRESS_LIST_NAME) + log.warning("Address-list '%s' пуст!", ADDRESS_LIST_NAME) else: - log.info("Адресов для проверки: %d → %s", len(ip_list), ip_list) + log.info("Адресов: %d", len(ip_list)) - # Добавляем новые хосты в state (без сброса существующих) + now = datetime.now() for ip, comment in ip_list: - if ip not in failure_count: - failure_count[ip] = 0 - host_down[ip] = False + if ip not in hosts_state: + hosts_state[ip] = { + "comment": comment, "status": "unknown", + "rtt": None, "down_since": None, + "last_seen": None, "last_notify": None, + } + else: + hosts_state[ip]["comment"] = comment + + # Параллельный пинг + ping_results = asyncio.run(ping_all_async(api, ip_list)) + last_check_time = datetime.now() + now = last_check_time - # Пингуем каждый адрес for ip, comment in ip_list: - label = f"{ip} ({comment})" if comment else ip - now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - rtt = mikrotik_ping(api, ip) + rtt = ping_results.get(ip) + state = hosts_state[ip] + label = fmt_label(ip) + ts = now.strftime("%Y-%m-%d %H:%M:%S") if rtt is not None: - log.info("✅ %-28s avg-rtt=%.1f ms", label, rtt) - if host_down[ip]: - host_down[ip] = False - failure_count[ip] = 0 + log.info("✅ %-34s %.1f ms", label, rtt) + state["rtt"] = rtt + state["last_seen"] = now + + if state["status"] == "down": + ds = state.get("down_since") + dur = fmt_duration(now - ds) if ds else "?" + state.update({"status": "up", "down_since": None, "last_notify": None}) notify( f"✅ Хост восстановился\n" - f"IP: {ip}" + (f" ({comment})" if comment else "") + f"\n" + f"IP: {ip}" + (f" ({comment})" if comment else "") + "\n" f"RTT: {rtt:.1f} ms\n" - f"Время: {now}" + f"Даунтайм: {dur}\n" + f"Время: {ts}" ) else: - failure_count[ip] = 0 + state["status"] = "up" + else: - failure_count[ip] += 1 - log.warning( - "❌ %-28s не отвечает (%d/%d)", - label, failure_count[ip], FAILURE_THRESHOLD, - ) - if failure_count[ip] >= FAILURE_THRESHOLD and not host_down[ip]: - host_down[ip] = True + log.warning("❌ %-34s нет ответа", label) + state["rtt"] = None + + if state["status"] != "down": + state.update({"status": "down", "down_since": now, "last_notify": now}) notify( f"🔴 Хост недоступен!\n" - f"IP: {ip}" + (f" ({comment})" if comment else "") + f"\n" + f"IP: {ip}" + (f" ({comment})" if comment else "") + "\n" f"Список: {ADDRESS_LIST_NAME}\n" - f"Не отвечает {failure_count[ip]} проверки подряд\n" - f"Время: {now}" + f"Время: {ts}" ) + elif REPEAT_NOTIFY_MIN > 0 and state.get("last_notify"): + mins = (now - state["last_notify"]).total_seconds() / 60 + if mins >= REPEAT_NOTIFY_MIN: + ds = state.get("down_since") + dur = fmt_duration(now - ds) if ds else "?" + state["last_notify"] = now + notify( + f"🔴 Хост всё ещё недоступен!\n" + f"IP: {ip}" + (f" ({comment})" if comment else "") + "\n" + f"Недоступен уже: {dur}\n" + f"Время: {ts}" + ) + + asyncio.run(process_bot_commands()) + check_daily_report() except Exception as exc: log.error("Ошибка: %s", exc) - notify(f"⚠️ Ошибка подключения к MikroTik\n{exc}") + notify( + f"⚠️ Ошибка подключения к MikroTik\n" + f"🔧 Роутер: {router_identity} ({MIKROTIK_HOST})\n" + f"{exc}" + ) finally: if pool: try: