import json from typing import Any, Dict, List import httpx from app.services.mikrotik.client_base import MikroTikClient def parse_bps(value: Any) -> float: if value is None: return 0.0 s = str(value).strip().lower().replace(" ", "") if s == "": return 0.0 try: return float(s) except ValueError: pass multipliers = {"bps": 1.0, "kbps": 1_000.0, "mbps": 1_000_000.0, "gbps": 1_000_000_000.0} for unit, mul in multipliers.items(): if s.endswith(unit): num = s[: -len(unit)] try: return float(num) * mul except ValueError: return 0.0 return 0.0 class MikroTikRESTClient(MikroTikClient): def __init__(self, base_url: str, username: str, password: str, verify_ssl: bool): self.base = base_url.rstrip("/") + "/rest" self.auth = (username, password) self.verify = verify_ssl async def _client(self) -> httpx.AsyncClient: return httpx.AsyncClient( base_url=self.base, auth=self.auth, verify=self.verify, timeout=httpx.Timeout(10.0), headers={"Content-Type": "application/json"}, ) async def list_interfaces(self) -> List[Dict[str, Any]]: params = {".proplist": "name,type,disabled,running"} async with await self._client() as c: r = await c.get("/interface", params=params) r.raise_for_status() data = r.json() if isinstance(data, dict): return [data] return data async def monitor_traffic_once(self, iface: str) -> Dict[str, Any]: payload = {"interface": iface, "once": ""} async with await self._client() as c: r = await c.post("/interface/monitor-traffic", content=json.dumps(payload)) r.raise_for_status() data = r.json() if isinstance(data, list) and data: return data[0] if isinstance(data, dict): return data return {}