fix: кодировка при чтении данных с MikroTik (кириллица в identity)

This commit is contained in:
2026-05-17 15:38:31 +07:00
parent 2d4307d2aa
commit da3bc4ee7a
+15 -9
View File
@@ -70,22 +70,26 @@ def notify(message: str) -> None:
# ────────────────────────────────────────────── # ──────────────────────────────────────────────
# MikroTik: получить список IP из address-list # 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]: def fetch_address_list(api) -> list[str]:
""" """
Читает /ip firewall address-list и возвращает список IP Читает /ip firewall address-list и возвращает список IP
из записей с именем ADDRESS_LIST_NAME (пропускает отключённые). из записей с именем ADDRESS_LIST_NAME (пропускает отключённые).
""" """
resource = api.get_resource("/ip/firewall/address-list") resource = api.get_binary_resource("/ip/firewall/address-list")
entries = resource.get(list=ADDRESS_LIST_NAME) entries = resource.get(**{b"list": ADDRESS_LIST_NAME.encode()})
ip_list = [] ip_list = []
for entry in entries: for entry in entries:
# Пропускаем отключённые записи if decode(entry.get(b"disabled", b"false")) == "true":
if entry.get("disabled", "false") == "true":
log.debug("Пропущена (disabled): %s", entry.get("address"))
continue continue
address = entry.get("address", "").strip() address = decode(entry.get(b"address", b"")).strip()
# Убираем префикс /32 если есть — пингуем чистый IP
if "/" in address: if "/" in address:
address = address.split("/")[0] address = address.split("/")[0]
if address: if address:
@@ -161,9 +165,11 @@ def mikrotik_ping(api, address: str) -> Optional[float]:
# ────────────────────────────────────────────── # ──────────────────────────────────────────────
def fetch_identity(api) -> str: def fetch_identity(api) -> str:
try: try:
resource = api.get_resource("/system/identity") resource = api.get_binary_resource("/system/identity")
result = resource.get() 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: except Exception:
return MIKROTIK_HOST return MIKROTIK_HOST