From da3bc4ee7ad1436cc8e676421ef0ff94da1a024a Mon Sep 17 00:00:00 2001
From: Sergey
Date: Sun, 17 May 2026 15:38:31 +0700
Subject: [PATCH] =?UTF-8?q?fix:=20=D0=BA=D0=BE=D0=B4=D0=B8=D1=80=D0=BE?=
=?UTF-8?q?=D0=B2=D0=BA=D0=B0=20=D0=BF=D1=80=D0=B8=20=D1=87=D1=82=D0=B5?=
=?UTF-8?q?=D0=BD=D0=B8=D0=B8=20=D0=B4=D0=B0=D0=BD=D0=BD=D1=8B=D1=85=20?=
=?UTF-8?q?=D1=81=20MikroTik=20(=D0=BA=D0=B8=D1=80=D0=B8=D0=BB=D0=BB=D0=B8?=
=?UTF-8?q?=D1=86=D0=B0=20=D0=B2=20identity)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
monitor.py | 24 +++++++++++++++---------
1 file changed, 15 insertions(+), 9 deletions(-)
diff --git a/monitor.py b/monitor.py
index 3e9289d..efe069b 100644
--- a/monitor.py
+++ b/monitor.py
@@ -70,22 +70,26 @@ def notify(message: str) -> None:
# ──────────────────────────────────────────────
# MikroTik: получить список IP из address-list
# ──────────────────────────────────────────────
+def decode(val) -> str:
+ """Декодирует bytes или str, игнорируя ошибки кодировки."""
+ if isinstance(val, bytes):
+ return val.decode("utf-8", errors="replace")
+ return str(val) if val else ""
+
+
def fetch_address_list(api) -> list[str]:
"""
Читает /ip firewall address-list и возвращает список IP
из записей с именем ADDRESS_LIST_NAME (пропускает отключённые).
"""
- resource = api.get_resource("/ip/firewall/address-list")
- entries = resource.get(list=ADDRESS_LIST_NAME)
+ resource = api.get_binary_resource("/ip/firewall/address-list")
+ entries = resource.get(**{b"list": ADDRESS_LIST_NAME.encode()})
ip_list = []
for entry in entries:
- # Пропускаем отключённые записи
- if entry.get("disabled", "false") == "true":
- log.debug("Пропущена (disabled): %s", entry.get("address"))
+ if decode(entry.get(b"disabled", b"false")) == "true":
continue
- address = entry.get("address", "").strip()
- # Убираем префикс /32 если есть — пингуем чистый IP
+ address = decode(entry.get(b"address", b"")).strip()
if "/" in address:
address = address.split("/")[0]
if address:
@@ -161,9 +165,11 @@ def mikrotik_ping(api, address: str) -> Optional[float]:
# ──────────────────────────────────────────────
def fetch_identity(api) -> str:
try:
- resource = api.get_resource("/system/identity")
+ resource = api.get_binary_resource("/system/identity")
result = resource.get()
- return result[0].get("name", MIKROTIK_HOST) if result else MIKROTIK_HOST
+ if result:
+ return decode(result[0].get(b"name", MIKROTIK_HOST.encode()))
+ return MIKROTIK_HOST
except Exception:
return MIKROTIK_HOST