#!/usr/bin/env python3 """ MikroTik IP Monitor — читает адреса из /ip firewall address-list, пингует параллельно, шлёт уведомления в Telegram и MAX (VK Teams). Возможности: - Параллельный пинг всех хостов - Повторные уведомления о даунтайме - Время даунтайма при восстановлении - Ежедневный отчёт - Команды /status, /help через Telegram и MAX - Веб-дашборд на порту 8080 """ import os import time import logging import asyncio 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")) MIKROTIK_USER = os.environ["MIKROTIK_USER"] MIKROTIK_PASSWORD = os.environ["MIKROTIK_PASSWORD"] 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_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_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")) # ────────────────────────────────────────────── # Логирование # ────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) 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 # ────────────────────────────────────────────── # Уведомления # ────────────────────────────────────────────── async def send_telegram(message: str) -> None: if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: return url = f"{TELEGRAM_API_URL}/bot{TELEGRAM_BOT_TOKEN}/sendMessage" payload = {"chat_id": TELEGRAM_CHAT_ID, "text": message, "parse_mode": "HTML"} async with httpx.AsyncClient(timeout=10) as client: try: resp = await client.post(url, data=payload) resp.raise_for_status() log.info("Telegram: отправлено") except httpx.HTTPStatusError as exc: log.error("Telegram: %s — %s", exc.response.status_code, exc.response.text) except Exception as exc: log.error("Telegram: %s", exc) async def send_max(message: str) -> None: if not MAX_BOT_TOKEN or not MAX_CHAT_ID: return 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={"chat_id": MAX_CHAT_ID}, json=payload, ) resp.raise_for_status() log.info("MAX: отправлено") except Exception as exc: log.error("MAX: %s", exc) def notify(message: str) -> None: async def _all(): await asyncio.gather(send_telegram(message), send_max(message)) asyncio.run(_all()) # ────────────────────────────────────────────── # MikroTik утилиты # ────────────────────────────────────────────── def decode(val) -> str: if isinstance(val, bytes): try: return val.decode("utf-8") except UnicodeDecodeError: return val.decode("windows-1251", errors="replace") return str(val) if val else "" def fetch_address_list(api) -> list[tuple[str, str]]: ip_list = [] 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 address = decode(entry.get("address", b"")).strip() if "/" in address: address = address.split("/")[0] comment = decode(entry.get("comment", b"")).strip() if address: ip_list.append((address, comment)) return ip_list def fetch_identity(api) -> str: try: 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: global router_identity, last_check_time 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: pool = routeros_api.RouterOsApiPool( MIKROTIK_HOST, username=MIKROTIK_USER, password=MIKROTIK_PASSWORD, port=MIKROTIK_PORT, plaintext_login=True, ) api = pool.get_api() log.info("Подключено к MikroTik") ip_list = fetch_address_list(api) if not startup_notified: 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\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) else: log.info("Адресов: %d", len(ip_list)) now = datetime.now() for ip, comment in ip_list: 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: 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("✅ %-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 "") + "\n" f"RTT: {rtt:.1f} ms\n" f"Даунтайм: {dur}\n" f"Время: {ts}" ) else: state["status"] = "up" else: 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 "") + "\n" f"Список: {ADDRESS_LIST_NAME}\n" 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" f"🔧 Роутер: {router_identity} ({MIKROTIK_HOST})\n" f"{exc}" ) finally: if pool: try: pool.disconnect() except Exception: pass log.info("Следующая проверка через %d сек …\n", CHECK_INTERVAL) time.sleep(CHECK_INTERVAL) if __name__ == "__main__": main()