ayurishchevandClaude Sonnet 5
4c1841b61b
Control Server (FastAPI) с WEB UI (Jinja2 + HTMX) и JSON API для группового администрирования устройств RouterOS 7 через REST. - Устройства: список, статус (модель, канал, версии ROS/FW, uptime, доступные обновления), примечания, пароли шифруются (Fernet); фоновый опрос каждые 30 с и быстрое обнаружение недоступности (таймаут соединения 4 с). - Группы устройств и фильтры (устройства, резервные копии). - Резервные копии: .backup и .rsc (show-sensitive) создаются через REST, скачиваются сервером и загружаются в S3 (Yandex Object Storage). - Обновление ROS/FW и выбор канала, групповые операции задачами. - Интерфейс по утверждённому макету: светлая/тёмная темы, окна, шрифты IBM Plex локально. Docker Compose, SQLite в томе, минимальный набор тестов. - Планы и итоги каждого изменения — в docs/changes/. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
69 lines
2.7 KiB
Python
69 lines
2.7 KiB
Python
import json as jsonlib
|
|
|
|
import httpx
|
|
|
|
|
|
class RosError(Exception):
|
|
"""Ошибка обращения к устройству RouterOS. status=None — сбой на уровне соединения."""
|
|
|
|
def __init__(self, message: str, status: int | None = None):
|
|
super().__init__(message)
|
|
self.status = status
|
|
|
|
|
|
class RosClient:
|
|
"""Тонкий клиент RouterOS REST API (https://<host>:<port>/rest/...)."""
|
|
|
|
def __init__(
|
|
self,
|
|
host: str,
|
|
port: int,
|
|
username: str,
|
|
password: str,
|
|
verify_tls: bool = False,
|
|
timeout: float = 30.0,
|
|
use_tls: bool = True,
|
|
transport: httpx.AsyncBaseTransport | None = None,
|
|
connect_timeout: float = 4.0,
|
|
):
|
|
self._connect = connect_timeout
|
|
self._http = httpx.AsyncClient(
|
|
base_url=f"{'https' if use_tls else 'http'}://{host}:{port}/rest",
|
|
auth=(username, password),
|
|
verify=verify_tls,
|
|
timeout=httpx.Timeout(timeout, connect=connect_timeout),
|
|
transport=transport,
|
|
)
|
|
|
|
async def __aenter__(self) -> "RosClient":
|
|
return self
|
|
|
|
async def __aexit__(self, *exc) -> None:
|
|
await self._http.aclose()
|
|
|
|
async def request(self, method: str, path: str, json: dict | None = None,
|
|
params: dict | None = None, timeout: float | None = None):
|
|
try:
|
|
kwargs = {"timeout": httpx.Timeout(timeout, connect=self._connect)} if timeout else {}
|
|
r = await self._http.request(method, path, json=json, params=params, **kwargs)
|
|
except httpx.HTTPError as e:
|
|
raise RosError(f"{self._http.base_url}: {type(e).__name__}: {e or repr(e.__cause__)}") from e
|
|
if r.status_code >= 400:
|
|
try:
|
|
body = jsonlib.loads(r.content.decode("utf-8", "replace"))
|
|
detail = body.get("detail") or body.get("message") or r.text
|
|
except ValueError:
|
|
detail = r.text
|
|
raise RosError(f"HTTP {r.status_code}: {detail}", status=r.status_code)
|
|
# RouterOS может отдавать имена файлов не в UTF-8 — декодируем с заменой, а не падаем
|
|
return jsonlib.loads(r.content.decode("utf-8", "replace")) if r.content else None
|
|
|
|
async def get(self, path: str, **params):
|
|
return await self.request("GET", path, params=params or None)
|
|
|
|
async def post(self, path: str, json: dict | None = None, timeout: float | None = None):
|
|
return await self.request("POST", path, json=json or {}, timeout=timeout)
|
|
|
|
async def delete(self, path: str):
|
|
return await self.request("DELETE", path)
|