From 1eb0ff1ecad17022c1186f0b6a315254b6a220b8 Mon Sep 17 00:00:00 2001 From: IluaAir Date: Wed, 25 Jun 2025 10:32:51 +0300 Subject: [PATCH] add oxiApi --- oxi/__init__.py | 0 oxi/interface/__init__.py | 12 +++---- oxi/interface/base.py | 6 ++-- oxi/interface/models/__init__.py | 7 ++++ oxi/interface/models/bdcom.py | 7 ++++ oxi/interface/models/mikrotik.py | 7 ++++ oxi/interface/models/qtech.py | 16 +++++++++ oxi/interface/models/vrp.py | 30 +++++++++++++++++ oxi/interface/registry.py | 12 +++++++ oxi/manager.py | 56 +++++++++++++++++++++----------- pynet.py | 9 +++-- 11 files changed, 129 insertions(+), 33 deletions(-) create mode 100644 oxi/__init__.py create mode 100644 oxi/interface/models/__init__.py create mode 100644 oxi/interface/models/bdcom.py create mode 100644 oxi/interface/models/mikrotik.py create mode 100644 oxi/interface/models/qtech.py create mode 100644 oxi/interface/models/vrp.py create mode 100644 oxi/interface/registry.py diff --git a/oxi/__init__.py b/oxi/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/oxi/interface/__init__.py b/oxi/interface/__init__.py index 8ce0323..147bb06 100644 --- a/oxi/interface/__init__.py +++ b/oxi/interface/__init__.py @@ -1,10 +1,6 @@ -from .vrp import Vrp -from .qtech import Qtech -from .mikrotik import Mikrotik -from .bdcom import BDcom +from oxi.interface.registry import register_parser, device_registry + __all__ = [ - 'Vrp', - 'Qtech', - 'Mikrotik', - 'BDcom' + 'register_parser', + 'device_registry' ] diff --git a/oxi/interface/base.py b/oxi/interface/base.py index 26dbfc8..ba40d5b 100644 --- a/oxi/interface/base.py +++ b/oxi/interface/base.py @@ -1,6 +1,6 @@ import re -from abc import ABC from dataclasses import dataclass +from typing import Protocol @dataclass @@ -25,7 +25,7 @@ class ParsedDeviceData: vlans: list[Vlan] -class BaseDevice(ABC): +class BaseDevice(Protocol): anchor_pattern: str = '!' hostname_pattern: str = 'hostname' @@ -81,7 +81,7 @@ class BaseDevice(ABC): ip_address_match = re.search(r'\s?ip address\s(.+)$', interface_block, re.MULTILINE) if not description_match and not ip_address_match: continue - ip_address = ip_address_match.group(1) if ip_address_match else None + ip_address = ip_address_match.group(1).replace(' ', '/') if ip_address_match else None description = description_match.group(1) if description_match else None interfaces.append(L3Interface( interface=interface_name, diff --git a/oxi/interface/models/__init__.py b/oxi/interface/models/__init__.py new file mode 100644 index 0000000..a3a9678 --- /dev/null +++ b/oxi/interface/models/__init__.py @@ -0,0 +1,7 @@ +import importlib +import pkgutil + +package = __package__ + +for loader, module_name, is_pkg in pkgutil.iter_modules(__path__): + importlib.import_module(f"{package}.{module_name}") diff --git a/oxi/interface/models/bdcom.py b/oxi/interface/models/bdcom.py new file mode 100644 index 0000000..dc7865a --- /dev/null +++ b/oxi/interface/models/bdcom.py @@ -0,0 +1,7 @@ +from oxi.interface import register_parser +from oxi.interface.models.qtech import Qtech + + +@register_parser("BDCOM") +class BDcom(Qtech): + pass diff --git a/oxi/interface/models/mikrotik.py b/oxi/interface/models/mikrotik.py new file mode 100644 index 0000000..623330e --- /dev/null +++ b/oxi/interface/models/mikrotik.py @@ -0,0 +1,7 @@ +from oxi.interface import register_parser +from oxi.interface.base import BaseDevice + + +@register_parser("Mikrotik") +class Mikrotik(BaseDevice): + ... diff --git a/oxi/interface/models/qtech.py b/oxi/interface/models/qtech.py new file mode 100644 index 0000000..1b8532e --- /dev/null +++ b/oxi/interface/models/qtech.py @@ -0,0 +1,16 @@ +import re + +from oxi.interface import register_parser +from oxi.interface.base import BaseDevice + + +@register_parser("QTECH") +class Qtech(BaseDevice): + + def __init__(self, config): + self.config: str = self._fix_config(config) + + def _fix_config(self, config): + pattern = r"Pending configurations.*" + cleaned_text = re.sub(pattern, "", config, flags=re.DOTALL) + return cleaned_text diff --git a/oxi/interface/models/vrp.py b/oxi/interface/models/vrp.py new file mode 100644 index 0000000..c7a02e0 --- /dev/null +++ b/oxi/interface/models/vrp.py @@ -0,0 +1,30 @@ +import re + +from oxi.interface import register_parser +from oxi.interface.base import BaseDevice, Vlan + + +@register_parser("VRP") +class Vrp(BaseDevice): + anchor_pattern: str = '#' + hostname_pattern = 'sysname' + unamed_vlan_splitter = ' ' + unamed_vlan_counter = 'to' + + def _parse_unamed_vlans(self) -> list[Vlan]: + vlans = [] + pattern = self.unamed_vlans_parse_pattern + for match in re.finditer(pattern, self.config, re.MULTILINE): + tokens = match.group(1).split(self.unamed_vlan_splitter) + i = 0 + while i < len(tokens): + if i + 2 < len(tokens) and tokens[i + 1].lower() == 'to': + start = int(tokens[i]) + end = int(tokens[i + 2]) + for vlan_id in range(start, end + 1): + vlans.append(Vlan(vlan=str(vlan_id), name=None, description=None)) + i += 3 # пропустить X, 'to', Y + else: + vlans.append(Vlan(vlan=str(tokens[i]), name=None, description=None)) + i += 1 + return vlans diff --git a/oxi/interface/registry.py b/oxi/interface/registry.py new file mode 100644 index 0000000..d4a0584 --- /dev/null +++ b/oxi/interface/registry.py @@ -0,0 +1,12 @@ +from typing import Callable, Type + +from oxi.interface.base import BaseDevice + +device_registry = {} + + +def register_parser(name: str) -> Callable[[Type[BaseDevice]], Type[BaseDevice]]: + def wrapper(cls): + device_registry[name.lower()] = cls + return cls + return wrapper diff --git a/oxi/manager.py b/oxi/manager.py index 0f93695..2a0cfb7 100644 --- a/oxi/manager.py +++ b/oxi/manager.py @@ -1,9 +1,13 @@ import logging +from functools import cached_property import requests from typing import Optional +from oxi.interface.base import BaseDevice from settings import settings +import oxi.interface.models # noqa +from oxi.interface import device_registry log = logging.getLogger() @@ -21,10 +25,7 @@ class OxidizedAPI: self._session.verify = verify if username and password: self._session.auth = (username, password) - - @property - def node(self): - return Node(self._session, self.base_url) + self.node = Node(self._session, self.base_url) def __enter__(self): return self @@ -48,15 +49,15 @@ class Node: self._base_url = base_url self._data = None - def show(self, name: str) -> 'NodeView': + def __call__(self, name: str) -> 'NodeView': url = f"{self._base_url}/node/show/{name}" if not url.endswith('.json'): url += '.json' - self._data = self._session.get(url) + response = self._session.get(url) return NodeView( session=self._session, base_url=self._base_url, - data=self._data.json() + data=response.json() ) @@ -84,26 +85,43 @@ class NodeView: @property def config(self): - return NodeConfig(self._session, self.full_name, self._base_url) + return NodeConfig(self._session, self.full_name, self.model, self._base_url) class NodeConfig: - def __init__(self, session: requests.Session, full_name: str, base_url: str): + def __init__(self, session: requests.Session, full_name: str, model: str, base_url: str): self._session = session self._full_name = full_name + self._model = model self._url = f"{base_url}/node/fetch/{full_name}" + self._device: type[BaseDevice] = device_registry.get(self._model.lower()) + if self._device is None: + raise ValueError(f"Device model '{self._model}' not found in registry") + self._data = None - def text(self) -> str: - return self._session.get(self._url).text + @cached_property + def _response(self): + log.debug(f"Fetching config from {self._url}") + response = self._session.get(self._url) + response.raise_for_status() + return response - def json(self) -> dict: - return self._session.get(self._url).json() + @property + def text(self): + return self._response.text - def __str__(self) -> str: - return self.text() + @property + def json(self): + return self._response.json() + def __str__(self): + return self.text -oxi = OxidizedAPI(username=settings.oxi_username, password=settings.oxi_password, verify=False) -print(oxi.node.show('Novok_HOME').config) -mikrotik = oxi.node.show('Novok_HOME').model -print(mikrotik.model) \ No newline at end of file + def vlans(self): + return self._device(self.text).parse_config().vlans + + def l3interfaces(self): + return self._device(self.text).parse_config().l3interfaces + + def vlaninterfaces(self): + return self._device(self.text).parse_config().vlaninterfaces diff --git a/pynet.py b/pynet.py index f902572..d736f3f 100644 --- a/pynet.py +++ b/pynet.py @@ -1,7 +1,9 @@ import pynetbox +from oxi.manager import OxidizedAPI from settings import settings + netbox = pynetbox.api( settings.nb_url, token=settings.nb_token) @@ -13,6 +15,7 @@ filters = { "role": "Kommutator", } -devices = netbox.dcim.devices.filter(**filters) -for device in devices: - print(f"{device.name} (IP: {device.primary_ip})") \ No newline at end of file + +# devices = netbox.dcim.devices.filter(**filters) +# for device in devices: +# print(f"{device.name} (IP: {device.primary_ip})") \ No newline at end of file