2394296f5b
- Implemented the `interfaces` and `vlans` methods in the `Keenetic` model to process and decode interface and VLAN data. - Added a `_decode_utf` method to handle UTF-8 encoded descriptions. - Updated the Keenetic TTP template to define structured groups for system, interfaces, and VLANs. - Refactored file paths in the `Mikrotik` model for consistency and clarity.
119 lines
4.0 KiB
Python
119 lines
4.0 KiB
Python
from abc import ABC, abstractmethod
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
from ttp import ttp
|
|
from oxi.interfaces.contract import Device
|
|
import xml.etree.ElementTree as ET
|
|
|
|
if TYPE_CHECKING:
|
|
from oxi.interfaces.contract import Interfaces, System, Vlans
|
|
|
|
|
|
class BaseDevice(ABC):
|
|
_REQUIRED_SECTIONS: frozenset[str] = frozenset({"system", "interfaces"})
|
|
_OPTIONAL_SECTIONS: frozenset[str] = frozenset({"vlans"})
|
|
|
|
def __init__(self, config: str):
|
|
self.config: str = config
|
|
self._loaded_template = self._load_template()
|
|
self._validate_template_groups()
|
|
self._raw: dict = self._run_ttp()
|
|
|
|
@property
|
|
@abstractmethod
|
|
def template(self) -> str:
|
|
"""
|
|
:return:
|
|
"""
|
|
|
|
@abstractmethod
|
|
def vlans(self) -> list["Vlans"]:
|
|
"""
|
|
Parse VLAN configuration from self._raw['vlans'].
|
|
|
|
Expected raw structure:
|
|
[{"id": 10, "description": "MGMT"}, {"id": 15, "name": "SSH"}, ...]
|
|
|
|
Returns:
|
|
list[Vlans]: список VLAN из секции vlans,
|
|
пустой список если секция отсутствует.
|
|
|
|
Raises:
|
|
ValueError: если _raw содержит некорректные данные.
|
|
"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def interfaces(self) -> list["Interfaces"]:
|
|
"""
|
|
Parse Interface configuration from self._raw['interfaces'].
|
|
|
|
Expected raw structure:
|
|
[{"name": "GEthernet1/0/1", "ip_address": "192.168.1.1", "mask": "24", "description": "IPBB interface"}]
|
|
|
|
Raises:
|
|
ValueError: если _raw содержит некорректные данные.
|
|
"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def system(self) -> "System":
|
|
"""
|
|
Parse System configuration from self._raw['system'].
|
|
|
|
Expected raw structure:
|
|
{"model":"RB951Ui-2nD", serial_number: "B88C0B31117B", "version": "7.12.1"}
|
|
|
|
Raises:
|
|
ValueError: если _raw содержит некорректные данные.
|
|
"""
|
|
...
|
|
|
|
def _load_template(self):
|
|
"""Подгрузка темплейтов из папки models/templates"""
|
|
path = Path(__file__).parent / "models" / "templates" / self.template
|
|
if not path.exists():
|
|
print("-" * 12)
|
|
print(path)
|
|
raise FileNotFoundError(f"Template {self.template} not found")
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
def _validate_template_groups(self) -> None:
|
|
"""Проверка что TTP темлпейт имеет декларированные группы для всех требуемых и опциональных секций"""
|
|
try:
|
|
root = ET.fromstring(self._loaded_template)
|
|
except ET.ParseError:
|
|
root = ET.fromstring(f"<template>{self._loaded_template}</template>")
|
|
|
|
declared = {g.get("name") for g in root.iter("group") if g.get("name")}
|
|
expected = self._REQUIRED_SECTIONS | self._OPTIONAL_SECTIONS
|
|
missing = expected - declared
|
|
|
|
if missing:
|
|
raise ValueError(
|
|
f"{self.__class__.__name__}: template '{self.template}' "
|
|
f"missing group declarations: {sorted(missing)}. "
|
|
f"Declared groups: {sorted(declared)}"
|
|
)
|
|
|
|
def _run_ttp(self) -> dict:
|
|
""" Основной парсер """
|
|
p = ttp(data=self.config, template=self._loaded_template)
|
|
p.parse()
|
|
raw: dict = p.result()[0][0]
|
|
missing = self._REQUIRED_SECTIONS - raw.keys()
|
|
if missing:
|
|
raise ValueError(
|
|
f"{self.__class__.__name__}: TTP template '{self.template}' "
|
|
f"did not produce required sections: {sorted(missing)}. "
|
|
f"Got: {(raw.keys())}"
|
|
)
|
|
return raw
|
|
|
|
def parse(self) -> Device:
|
|
return Device(
|
|
system=self.system(),
|
|
interfaces=self.interfaces(),
|
|
vlans=self.vlans(),
|
|
)
|