138 lines
5.2 KiB
Python
138 lines
5.2 KiB
Python
import ipaddress
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Protocol
|
||
|
||
|
||
@dataclass
|
||
class L3Interface:
|
||
interface: str
|
||
description: str | None = None
|
||
ip_address: str | None = None
|
||
|
||
|
||
@dataclass
|
||
class Vlan:
|
||
vlan: int
|
||
name: str | None
|
||
description: str | None
|
||
|
||
|
||
@dataclass
|
||
class ParsedDeviceData:
|
||
hostname: str
|
||
l3interfaces: list[L3Interface]
|
||
vlaninterfaces: list[L3Interface]
|
||
vlans: list[Vlan]
|
||
|
||
|
||
class BaseDevice(Protocol):
|
||
anchor_pattern: str = '!'
|
||
hostname_pattern: str = 'hostname'
|
||
|
||
@property
|
||
def l3interface_parse_pattern(self):
|
||
return rf"^interface([^\n]*)\n(.*?)(?=^{self.anchor_pattern})"
|
||
|
||
@property
|
||
def vlan_parse_pattern(self):
|
||
return rf"^vlan\s+(\d{{1,4}})\r?\n(.*?)(?=^{self.anchor_pattern}|\Z)"
|
||
|
||
unamed_vlan_splitter: str = ','
|
||
unamed_vlan_counter = '-'
|
||
unamed_vlans_parse_pattern = r"vlan\s+(?:\w+\s+)?([\d,-]*[ ,][\d (to),-]*)$"
|
||
|
||
def __init__(self, config):
|
||
self.config: str = config
|
||
|
||
def parse_config(self) -> ParsedDeviceData:
|
||
"""Парсит конфигурацию и возвращает структурированные данные."""
|
||
hostname = self._parse_hostname()
|
||
l3interfaces = self._parse_l3_interfaces()
|
||
vlaninterfaces = self._parse_vlan_interfaces(l3interfaces)
|
||
vlans = self._parse_vlans()
|
||
unamed_vlans = self._parse_unamed_vlans()
|
||
vlan_map: dict[int, Vlan] = {}
|
||
for v in unamed_vlans:
|
||
vlan_map[v.vlan] = v
|
||
for v in vlans:
|
||
vlan_map[v.vlan] = v
|
||
vlans = list(vlan_map.values())
|
||
return ParsedDeviceData(
|
||
hostname=hostname,
|
||
l3interfaces=l3interfaces,
|
||
vlaninterfaces=vlaninterfaces,
|
||
vlans=vlans
|
||
)
|
||
|
||
def _parse_hostname(self) -> str:
|
||
"""Извлекает hostname из конфигурации."""
|
||
pattern = self.hostname_pattern
|
||
match = re.search(rf"^{pattern}\s+(\S+)", self.config, re.MULTILINE)
|
||
return match.group(1) if match else "unknown"
|
||
|
||
def _parse_l3_interfaces(self) -> list[L3Interface]:
|
||
"""Парсит L3 интерфейсы"""
|
||
interfaces = []
|
||
pattern = self.l3interface_parse_pattern
|
||
for match in re.finditer(pattern, self.config, re.MULTILINE | re.DOTALL):
|
||
interface_name = match.group(1).replace(' ', '')
|
||
interface_block = match.group(2)
|
||
description_match = re.search(r'\s?description\s(.+)$', interface_block, re.MULTILINE)
|
||
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).replace(' ', '/').replace('\r', '') if ip_address_match else None
|
||
try:
|
||
if ip_address is not None:
|
||
ip_address = ipaddress.ip_interface(ip_address)
|
||
except ValueError:
|
||
continue
|
||
description = description_match.group(1) if description_match else None
|
||
interfaces.append(L3Interface(
|
||
interface=interface_name,
|
||
description=description,
|
||
ip_address=ip_address
|
||
))
|
||
return interfaces
|
||
|
||
@staticmethod
|
||
def _parse_vlan_interfaces(l3_interfaces: list[L3Interface]):
|
||
'''Парсит Vlan интерфейсы от уже собранный l3 интерфейсов'''
|
||
interfaces = []
|
||
for interface in l3_interfaces:
|
||
if interface.interface.lower().startswith('vlan'):
|
||
interfaces.append(interface)
|
||
return interfaces
|
||
|
||
def _parse_vlans(self):
|
||
'''Парсит Vlan с naming'''
|
||
vlans = []
|
||
pattern = self.vlan_parse_pattern
|
||
for match in re.finditer(pattern, self.config, re.MULTILINE | re.DOTALL):
|
||
if match:
|
||
vlan = match.group(1)
|
||
vlan_block = match.group(2)
|
||
description_match = re.search(r'\s?description\s(.+)$', vlan_block, re.MULTILINE)
|
||
name_match = re.search(r'\s?name\s(.+)$', vlan_block, re.MULTILINE)
|
||
description = description_match.group(1) if description_match else None
|
||
name = name_match.group(1) if name_match else None
|
||
vlans.append(Vlan(vlan=vlan, description=description, name=name))
|
||
return vlans
|
||
|
||
def _parse_unamed_vlans(self) -> list[Vlan]:
|
||
'''Парсит строчку с перечислением vlan'ов'''
|
||
vlans = []
|
||
pattern = self.unamed_vlans_parse_pattern
|
||
for match in re.finditer(pattern, self.config, re.MULTILINE):
|
||
if match:
|
||
_iter = match.group(1).split(self.unamed_vlan_splitter)
|
||
for part in _iter:
|
||
if self.unamed_vlan_counter in part:
|
||
start, end = map(int, part.split(self.unamed_vlan_counter))
|
||
for vlan_id in range(start, end + 1):
|
||
vlans.append(Vlan(vlan=str(vlan_id), name=None, description=None))
|
||
else:
|
||
vlans.append(Vlan(vlan=str(part), name=None, description=None))
|
||
return vlans
|