diff --git a/.gitignore b/.gitignore index f0ef48f..4414f4f 100644 --- a/.gitignore +++ b/.gitignore @@ -188,7 +188,7 @@ cython_debug/ # that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore # and can be added to the global gitignore or merged into this file. However, if you prefer, # you could uncomment the following to ignore the entire vscode folder -# .vscode/ +.vscode/ # Ruff stuff: .ruff_cache/ diff --git a/.vscode/settings.json b/.vscode/settings.json deleted file mode 100644 index 69b361b..0000000 --- a/.vscode/settings.json +++ /dev/null @@ -1,3 +0,0 @@ -{ - "nixEnvSelector.nixFile": "${workspaceFolder}/flake.nix" -} \ No newline at end of file diff --git a/UML/structure.jpg b/UML/structure.jpg deleted file mode 100644 index 8f33bdb..0000000 Binary files a/UML/structure.jpg and /dev/null differ diff --git a/data/input/1НФ VLAN - 1НФ.csv b/data/input/1НФ VLAN - 1НФ.csv deleted file mode 100644 index f3b8583..0000000 --- a/data/input/1НФ VLAN - 1НФ.csv +++ /dev/null @@ -1,11 +0,0 @@ -Name,Role,Interface,Network,VLAN,Network IP,Mask,Device IP,Default Gateway -PC1,Host,eth1,A,4,10.0.0.0,/24,10.0.0.1,0.0.0.0 -PC2,Host,eth1,B,2,10.0.0.0,/24,10.0.0.2,0.0.0.0 -PC3,Host,eth1,C,3,10.0.0.0,/24,10.0.0.3,0.0.0.0 -PC4,Host,eth1,D,4,10.0.0.0,/24,10.0.0.4,0.0.0.0 -com_left,Switch,eth1,tr,trunk,,,, -com_left,Switch,eth2,A,4,,,, -com_left,Switch,eth3,B,2,,,, -com_right,Switch,eth1,tr,trunk,,,, -com_right,Switch,eth2,D,4,,,, -com_right,Switch,eth3,C,3,,,, \ No newline at end of file diff --git a/data/input/host-host.csv b/data/input/host-host.csv deleted file mode 100644 index 94a6413..0000000 --- a/data/input/host-host.csv +++ /dev/null @@ -1,6 +0,0 @@ -name,"type","interfaces","ip","endpoints" -PC1,"host","int1","","" -PC2,"host","int2","","" -int1,"interface","","10.0.12.1/24","" -int2,"interface","","10.0.12.2/24","" -link,"wire","","","int1,int2" \ No newline at end of file diff --git a/data/input/table.csv b/data/input/table.csv deleted file mode 100644 index f3b8583..0000000 --- a/data/input/table.csv +++ /dev/null @@ -1,11 +0,0 @@ -Name,Role,Interface,Network,VLAN,Network IP,Mask,Device IP,Default Gateway -PC1,Host,eth1,A,4,10.0.0.0,/24,10.0.0.1,0.0.0.0 -PC2,Host,eth1,B,2,10.0.0.0,/24,10.0.0.2,0.0.0.0 -PC3,Host,eth1,C,3,10.0.0.0,/24,10.0.0.3,0.0.0.0 -PC4,Host,eth1,D,4,10.0.0.0,/24,10.0.0.4,0.0.0.0 -com_left,Switch,eth1,tr,trunk,,,, -com_left,Switch,eth2,A,4,,,, -com_left,Switch,eth3,B,2,,,, -com_right,Switch,eth1,tr,trunk,,,, -com_right,Switch,eth2,D,4,,,, -com_right,Switch,eth3,C,3,,,, \ No newline at end of file diff --git a/data/output/diagram.png b/data/output/diagram.png deleted file mode 100644 index 27959a1..0000000 Binary files a/data/output/diagram.png and /dev/null differ diff --git a/data/output/output.yaml b/data/output/output.yaml deleted file mode 100644 index 45278eb..0000000 --- a/data/output/output.yaml +++ /dev/null @@ -1,19 +0,0 @@ -links: -networks: - - A: [PC1.eth0, PC2.eth0] -meta: - id: host-host.csv - name: host-host.csv -nodes: -- role: host - name: PC1 - interfaces: - - eth0: - - ip: 10.0.12.1/24 - network: A -- role: host - name: PC2 - interfaces: - - eth0: - - ip: 10.0.12.2/24 - network: A diff --git a/data/output/schema.yaml b/data/output/schema.yaml deleted file mode 100644 index e47ba5b..0000000 --- a/data/output/schema.yaml +++ /dev/null @@ -1,21 +0,0 @@ -meta: - id: "lab-02-basic-utils" - name: "Basic Network Utilities & Traffic Monitoring" - description: "Two hosts connected directly to each other." - -links: - - endpoints: ["PC1", "PC2"] - -nodes: - - name: PC1 - role: host - interfaces: - - ip: "10.0.12.1/24" - - - name: PC2 - role: host - interfaces: - - ip: "10.0.12.2/24" - - # Optional: - # - "ip link set dev eth1 down" diff --git a/data/output/table.yaml b/data/output/table.yaml deleted file mode 100644 index 8eababb..0000000 --- a/data/output/table.yaml +++ /dev/null @@ -1,81 +0,0 @@ -links: [] -networks: -- A: - members: - - PC1.eth1 - - com_left.eth2 - network_ip: 10.0.0.0 - mask: /24 - vlan: 4 -- B: - members: - - PC2.eth1 - - com_left.eth3 - network_ip: 10.0.0.0 - mask: /24 - vlan: 2 -- C: - members: - - PC3.eth1 - - com_right.eth3 - network_ip: 10.0.0.0 - mask: /24 - vlan: 3 -- D: - members: - - PC4.eth1 - - com_right.eth2 - network_ip: 10.0.0.0 - mask: /24 - vlan: 4 -- tr: - members: - - com_left.eth1 - - com_right.eth1 - vlan: trunk -meta: - id: table.csv - name: table.csv -nodes: -- role: host - name: PC1 - interfaces: - - eth1: - - ip: 10.0.0.1/24 - network: A -- role: host - name: PC2 - interfaces: - - eth1: - - ip: 10.0.0.2/24 - network: B -- role: host - name: PC3 - interfaces: - - eth1: - - ip: 10.0.0.3/24 - network: C -- role: host - name: PC4 - interfaces: - - eth1: - - ip: 10.0.0.4/24 - network: D -- role: switch - name: com_left - interfaces: - - eth1: - - network: tr - - eth2: - - network: A - - eth3: - - network: B -- role: switch - name: com_right - interfaces: - - eth1: - - network: tr - - eth2: - - network: D - - eth3: - - network: C diff --git a/src_old/check_correct.py b/src_old/check_correct.py deleted file mode 100755 index f417406..0000000 --- a/src_old/check_correct.py +++ /dev/null @@ -1,22 +0,0 @@ -from parse import parse_csv_to_structure -import sys -from structures import Host, Wire - - -def check_correct(data): - assert isinstance(data, dict) - - for key, value in data.items(): - assert isinstance(key, str) - assert isinstance(value, dict) - - # for subkey, subvalue in value.items(): - # assert isinstance(subkey, str) - # assert isinstance(subvalue, str) - - -if __name__ == "__main__": - data = parse_csv_to_structure(sys.argv[1]) - if check_correct(data): - print("yay!") - print(data) diff --git a/src_old/make_res.py b/src_old/make_res.py deleted file mode 100644 index 1ce5ec9..0000000 --- a/src_old/make_res.py +++ /dev/null @@ -1,251 +0,0 @@ -# export_yaml.py -from __future__ import annotations - -import argparse -from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple, Union - -from parse import parse_csv_topology - - -def _nonempty(v: Any) -> bool: - if v is None: - return False - if isinstance(v, str) and not v.strip(): - return False - return True - - -def _mask_to_prefix(mask: Optional[str]) -> str: - """ - Маска в CSV может быть '/24' или '24' или None. - Для записи ip/prefix нужно вернуть '24' (без слэша) либо ''. - """ - if not mask: - return "" - m = str(mask).strip() - if not m: - return "" - return m[1:] if m.startswith("/") else m - - -def _ip_with_prefix(device_ip: str, mask: Optional[str]) -> str: - ip = (device_ip or "").strip() - if not ip: - return "" - pfx = _mask_to_prefix(mask) - return f"{ip}/{pfx}" if pfx else ip - - -def _role_from_raw( - devices_raw: Dict[str, Dict[str, Any]], device_name: str, fallback: str -) -> str: - raw = devices_raw.get(device_name, {}) - role = str(raw.get("role", "") or "").strip().lower() - return role if role else fallback - - -def _network_vlan_value( - result_devices_raw: Dict[str, Dict[str, Any]], net_name: str -) -> Optional[Union[int, str]]: - """ - VLAN — свойство сети. Берём: - 1) если у Network-объекта есть vlan_id — используем int (обрабатывается выше) - 2) иначе пытаемся найти vlan_raw (строковый VLAN, например 'trunk') в devices_raw - """ - found_vlan_raw: Optional[str] = None - for d in result_devices_raw.values(): - for iface in d.get("interfaces", []): - nf = iface.get("network") or {} - if nf.get("name") == net_name and _nonempty(nf.get("vlan_raw")): - found_vlan_raw = str(nf.get("vlan_raw")).strip() - break - if found_vlan_raw: - break - return found_vlan_raw if _nonempty(found_vlan_raw) else None - - -# --- YAML dump (PyYAML preferred, fallback if not installed) --- -def _dump_yaml_fallback(obj: Any, indent: int = 0) -> str: - sp = " " * indent - - if obj is None: - return "null" - if isinstance(obj, bool): - return "true" if obj else "false" - if isinstance(obj, (int, float)): - return str(obj) - if isinstance(obj, str): - s = obj - if ( - s == "" - or s.strip() != s - or any(c in s for c in [":", "#", "{", "}", "[", "]"]) - ): - return f'"{s.replace(chr(34), r"\"")}"' - return s - - if isinstance(obj, list): - if not obj: - return "[]" - lines: List[str] = [] - for item in obj: - val = _dump_yaml_fallback(item, indent + 1) - if "\n" in val: - first, *rest = val.splitlines() - lines.append(f"{sp}- {first}") - for r in rest: - lines.append(f"{' ' * (indent + 1)}{r}") - else: - lines.append(f"{sp}- {val}") - return "\n".join(lines) - - if isinstance(obj, dict): - if not obj: - return "{}" - lines: List[str] = [] - for k, v in obj.items(): - key = str(k) - if isinstance(v, (dict, list)) and v: - lines.append(f"{sp}{key}:") - lines.append(_dump_yaml_fallback(v, indent + 1)) - else: - lines.append(f"{sp}{key}: {_dump_yaml_fallback(v, indent + 1)}") - return "\n".join(lines) - - return _dump_yaml_fallback(str(obj), indent) - - -def dump_yaml(obj: Any) -> str: - try: - import yaml # type: ignore - - return yaml.safe_dump(obj, sort_keys=False, allow_unicode=True) - except Exception: - return _dump_yaml_fallback(obj) + "\n" - - -def build_yaml_document(csv_path: Union[str, Path]) -> Dict[str, Any]: - csv_path = Path(csv_path) - result = parse_csv_topology(csv_path) - - doc: Dict[str, Any] = { - "links": [], - "networks": [], - "meta": {"id": csv_path.name, "name": csv_path.name}, - "nodes": [], - } - - # ----------------- - # NETWORKS (только тут: vlan, network_ip, mask) - # ----------------- - networks_block: List[Dict[str, Any]] = [] - for net_name in sorted(result.network_interfaces.keys()): - members = [ - f"{dev}.{iface}" for dev, iface in result.network_interfaces[net_name] - ] - - net_entry: Dict[str, Any] = {"members": members} - - net_obj = result.networks.get(net_name) - if net_obj is not None: - # network_ip, mask - if _nonempty(net_obj.ip): - net_entry["network_ip"] = net_obj.ip - if _nonempty(net_obj.subnet_mask): - net_entry["mask"] = net_obj.subnet_mask - - # vlan (int) - if _nonempty(net_obj.vlan_id): - net_entry["vlan"] = net_obj.vlan_id - - # vlan (string), если vlan_id отсутствует, но в raw есть vlan_raw - if "vlan" not in net_entry: - vlan_raw = _network_vlan_value(result.devices_raw, net_name) - if _nonempty(vlan_raw): - net_entry["vlan"] = vlan_raw - - networks_block.append({net_name: net_entry}) - - doc["networks"] = networks_block - - # ----------------- - # NODES (интерфейсы: без vlan/network_ip/mask) - # ----------------- - nodes_block: List[Dict[str, Any]] = [] - for dev_name in sorted(result.devices.keys()): - device = result.devices[dev_name] - role_fallback = type(device).__name__.lower() - role = _role_from_raw(result.devices_raw, dev_name, role_fallback) - - node: Dict[str, Any] = { - "role": role, - "name": device.name, - "interfaces": [], - } - - for iface in device.interfaces: - iface_items: List[Dict[str, Any]] = [] - - network_name = iface.network.name if iface.network else None - mask = iface.network.subnet_mask if iface.network else None - - # Если есть IP-адреса — пишем ip/prefix + default_gateway + network (+ mode при наличии) - if iface.ips: - for ip_obj in iface.ips: - entry: Dict[str, Any] = {} - - ip_comp = _ip_with_prefix(ip_obj.ip_str, mask) - if _nonempty(ip_comp): - entry["ip"] = ip_comp - - if _nonempty(network_name): - entry["network"] = network_name - - dg = getattr(ip_obj, "default_gateway", None) - if _nonempty(dg): - entry["default_gateway"] = dg - - if _nonempty(iface.mode): - entry["mode"] = iface.mode - - # добавляем только если есть хоть что-то - if entry: - iface_items.append(entry) - - else: - # Нет IP — оставим network/mode (например для switch портов) - entry: Dict[str, Any] = {} - if _nonempty(network_name): - entry["network"] = network_name - if _nonempty(iface.mode): - entry["mode"] = iface.mode - if entry: - iface_items.append(entry) - - node["interfaces"].append({iface.name: iface_items}) - - nodes_block.append(node) - - doc["nodes"] = nodes_block - return doc - - -def main() -> None: - ap = argparse.ArgumentParser( - description="Export topology parsed from CSV to YAML (network props only under networks)" - ) - ap.add_argument("csv", help="Path to input CSV") - ap.add_argument("-o", "--out", help="Path to output YAML (default: .yaml)") - args = ap.parse_args() - - csv_path = Path(args.csv) - out_path = Path(args.out) if args.out else csv_path.with_suffix(".yaml") - - doc = build_yaml_document(csv_path) - out_path.write_text(dump_yaml(doc), encoding="utf-8") - print(f"Saved: {out_path}") - - -if __name__ == "__main__": - main() diff --git a/src_old/old/structures copy.py b/src_old/old/structures copy.py deleted file mode 100644 index 48df349..0000000 --- a/src_old/old/structures copy.py +++ /dev/null @@ -1,61 +0,0 @@ -from dataclasses import dataclass, field -from typing import Dict, Any - - -@dataclass -class Device: - name: str - model: str - properties: Dict[str, Any] = field(default_factory=dict) - - def __post_init__(self): - if not self.name: - raise ValueError("Device name cannot be empty") - if not self.model: - raise ValueError("Device model cannot be empty") - if not isinstance(self.properties, dict): - raise TypeError("properties must be a dict") - - -@dataclass -class Switch(Device): - def __post_init__(self): - super().__post_init__() - - if "ports" not in self.properties: - raise ValueError(f"Switch '{self.name}' must define 'ports'") - if ( - not isinstance(self.properties["ports"], int) - or self.properties["ports"] <= 0 - ): - raise ValueError("'ports' must be a positive integer") - - -@dataclass -class Router(Device): - def __post_init__(self): - super().__post_init__() - - if "interfaces" not in self.properties: - raise ValueError(f"Router '{self.name}' must define 'interfaces'") - if ( - not isinstance(self.properties["interfaces"], int) - or self.properties["interfaces"] <= 0 - ): - raise ValueError("'interfaces' must be a positive integer") - - -@dataclass -class Host(Device): - def __post_init__(self): - super().__post_init__() - - ip = self.properties.get("ip") - if not ip or not isinstance(ip, str): - raise ValueError(f"Host '{self.name}' must contain a valid 'ip' string") - - -@dataclass -class Wire: - from_device: Device - to_device: Device diff --git a/src_old/old/structures.py b/src_old/old/structures.py deleted file mode 100644 index 8b239b6..0000000 --- a/src_old/old/structures.py +++ /dev/null @@ -1,182 +0,0 @@ -from dataclasses import dataclass, field -from typing import List, Optional, Dict - - -# =========================== -# Interface Classes -# =========================== - - -@dataclass -class Interface: - name: str - ip_address: Optional[str] = None - mac_address: Optional[str] = None - speed: Optional[str] = None - duplex: Optional[str] = None - - def __init__(self, fields): - self.name = fields.get("name") - self.ip_address = fields.get("ip") - self.mac_address = fields.get("mac") - self.speed = fields.get("speed") - self.duplex = fields.get("duplex") - self.__post_init__() - - def __post_init__(self): - if not isinstance(self.name, str) or not self.name: - raise ValueError("Interface 'name' must be a non-empty string") - - -@dataclass -class VirtualInterface(Interface): - vlan_id: int = 0 - parent_physical: Optional[Interface] = None - - def __init__(self, fields): - super().__init__(fields) - self.vlan_id = fields.get("vlanID", fields.get("vlan_id", 0)) - self.parent_physical = fields.get( - "parent_physical", fields.get("parentPhysical", fields.get("parent")) - ) - self.__post_init__() - - def __post_init__(self): - super().__post_init__() - if not isinstance(self.vlan_id, int) or self.vlan_id < 0: - raise ValueError( - "VirtualInterface 'vlan_id' must be a non-negative integer" - ) - if not isinstance(self.parent_physical, Interface): - raise TypeError( - "VirtualInterface 'parent_physical' must reference an Interface" - ) - - -# =========================== -# Network / Subnet -# =========================== - - -@dataclass -class Network: - cidr: str - gateway: Optional[str] = None - interfaces: List[Interface] = field(default_factory=list) - - def __init__(self, fields): - self.cidr = fields.get("cidr") - self.gateway = fields.get("gateway") - interfaces = fields.get("interfaces", []) - if isinstance(interfaces, str): - interfaces = [iface for iface in interfaces.split(",") if iface] - self.interfaces = interfaces - self.__post_init__() - - def __post_init__(self): - if not isinstance(self.cidr, str) or not self.cidr: - raise ValueError("Network 'cidr' must be a non-empty string") - if self.gateway is not None and not isinstance(self.gateway, str): - raise ValueError("Network 'gateway' must be a string") - - -# =========================== -# Device Base Class (Abstract) -# =========================== - - -@dataclass -class Device: - name: str - mgmt_ip: Optional[str] = None - interfaces: List[Interface] = field(default_factory=list) - - def __init__(self, fields): - self.name = fields.get("name") - self.mgmt_ip = fields.get("mgmtIP") - self.interfaces = [f for f in fields.get("interfaces").split(",") if f] - self.__post_init__() - - def __post_init__(self): - if not isinstance(self.name, str) or not self.name: - raise ValueError("Device 'name' must be a non-empty string") - if self.mgmt_ip is not None and not isinstance(self.mgmt_ip, str): - raise ValueError("Device 'mgmtIP' must be a string") - - -# =========================== -# Host / Router / Switch -# =========================== - - -@dataclass -class Host(Device): - operating_system: Optional[str] = None - - def __init__(self, fields): - super().__init__(fields) - self.operating_system = fields.get("operatingSystem") - self.__post_init__() - - def __post_init__(self): - super().__post_init__() - if not isinstance(self.operating_system, str): - raise ValueError("Host 'operatingSystem' must be a string") - - -@dataclass -class Router(Device): - routing_table: Dict[str, str] = field(default_factory=dict) - routing_protocols: List[str] = field(default_factory=list) - - def __init__(self, fields): - super().__init__(fields) - self.routing_table = fields.get("routingTable", {}) - self.routing_protocols = fields.get("routingProtocols", []) - self.__post_init__() - - def __post_init__(self): - super().__post_init__() - - -@dataclass -class Switch(Device): - mac_table: Dict[str, str] = field(default_factory=dict) - vlan_database: Dict[int, str] = field(default_factory=dict) - - def __init__(self, fields): - super().__init__(fields) - self.mac_table = fields.get("macTable", {}) - self.vlan_database = fields.get("vlanDatabase", {}) - self.__post_init__() - - def __post_init__(self): - super().__post_init__() - - -# =========================== -# Physical Link (Wire) -# =========================== - - -@dataclass -class Wire: - id: str - endpoints: List[Interface] # Must contain exactly 2 interfaces - bandwidth: Optional[str] = None - - def __init__(self, fields): - self.id = fields.get("name") - self.endpoints = [ep for ep in fields.get("endpoints").split(",") if ep] - self.bandwidth = fields.get("bandwidth") - self.__post_init__() - - def __post_init__(self): - if not isinstance(self.id, str) or not self.id: - raise ValueError("Wire 'id' must be a non-empty string") - if len(self.endpoints) != 2: - raise ValueError( - "Wire 'endpoints' must contain exactly 2 Interface objects" - ) - if not all(isinstance(i, Interface) for i in self.endpoints): - raise TypeError("Wire 'endpoints' must be Interface objects") diff --git a/src_old/parse.py b/src_old/parse.py deleted file mode 100644 index 6cf65ec..0000000 --- a/src_old/parse.py +++ /dev/null @@ -1,211 +0,0 @@ -# parse.py -from __future__ import annotations - -import csv -from dataclasses import dataclass -from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple, Union - -from structures import Device, Host, Router, Switch, Network, Interface - - -# =========================== -# CSV -> canonical fields -# =========================== -COLUMN_MAP: Dict[str, str] = { - "Name": "device_name", - "Role": "role", - "Interface": "interface_name", - "Network": "network_name", - "VLAN": "vlan", - "Network IP": "network_ip", - "Mask": "mask", - "Device IP": "device_ip", - "Default Gateway": "default_gateway", -} - -ROLE_MAP = { - "host": Host, - "switch": Switch, - "router": Router, -} - - -def _clean(v: Any) -> str: - return str(v).strip() if v is not None else "" - - -def _to_int(v: str) -> Optional[int]: - v = _clean(v) - if v.isdigit(): - return int(v) - return None - - -def _gateway_or_none(v: str) -> Optional[str]: - v = _clean(v) - if not v or v == "0.0.0.0": - return None - return v - - -def _infer_mode(network_name: str, vlan: str) -> str: - """ - Принятое допущение под ваш пример: - - если VLAN == 'trunk' или Network == 'tr'/'trunk' => mode='trunk' - - иначе => mode='access' - """ - n = _clean(network_name).lower() - v = _clean(vlan).lower() - if v == "trunk" or n in {"tr", "trunk"}: - return "trunk" - return "access" - - -@dataclass -class ParseResult: - # словарь, пригодный для сериализации/отладки и последующего создания структур - devices_raw: Dict[str, Dict[str, Any]] - # готовые python-объекты ваших классов Host/Switch/Router - devices: Dict[str, Device] - # агрегированные сети (по имени сети) - networks: Dict[str, Network] - # Network -> список (device_name, interface_name) - network_interfaces: Dict[str, List[Tuple[str, str]]] - - -def parse_csv_topology(path: Union[str, Path], encoding: str = "utf-8") -> ParseResult: - path = Path(path) - - devices_raw: Dict[str, Dict[str, Any]] = {} - network_interfaces: Dict[str, List[Tuple[str, str]]] = {} - network_agg: Dict[str, Dict[str, Any]] = {} - - with path.open(newline="", encoding=encoding) as f: - reader = csv.DictReader(f) - if not reader.fieldnames: - raise ValueError("CSV is empty or has no header row") - - for row in reader: - # 1) rename columns -> canonical keys - canon: Dict[str, str] = {} - for k, v in row.items(): - kk = COLUMN_MAP.get( - k, k - ) # если встретили неизвестную колонку — оставим как есть - canon[kk] = _clean(v) - - device_name = canon.get("device_name", "") - role = canon.get("role", "") - iface_name = canon.get("interface_name", "") - network_name = canon.get("network_name", "") - vlan_raw = canon.get("vlan", "") - net_ip = canon.get("network_ip", "") - mask = canon.get("mask", "") - dev_ip = canon.get("device_ip", "") - gw = canon.get("default_gateway", "") - - if not device_name: - continue - - # 2) init device record - dev_rec = devices_raw.setdefault( - device_name, - {"name": device_name, "role": role, "interfaces": []}, - ) - - # если роль у устройства не была проставлена ранее, но появилась сейчас — заполним - if role and not dev_rec.get("role"): - dev_rec["role"] = role - - # если интерфейс не указан — дальше нечего собирать - if not iface_name: - continue - - mode = _infer_mode(network_name, vlan_raw) - - # 3) network fields (для интерфейса) + агрегирование сетей - network_fields: Optional[Dict[str, Any]] = None - - if network_name: - vlan_id = _to_int(vlan_raw) # trunk -> None - network_fields = { - "name": network_name, - "vlan_id": vlan_id, # для структур (Network берёт vlan_id) - "vlan_raw": vlan_raw or None, # для YAML/отладки (как в CSV) - "ip": net_ip or None, # Network IP - "mask": mask or None, # Mask - } - - # агрегируем сеть: берём первое непустое значение - agg = network_agg.setdefault(network_name, {"name": network_name}) - if vlan_id is not None and agg.get("vlan_id") is None: - agg["vlan_id"] = vlan_id - if vlan_raw and not agg.get("vlan_raw"): - agg["vlan_raw"] = vlan_raw - if net_ip and not agg.get("ip"): - agg["ip"] = net_ip - if mask and not agg.get("mask"): - agg["mask"] = mask - - network_interfaces.setdefault(network_name, []).append( - (device_name, iface_name) - ) - - # 5) ips list - ips: List[Interface.IP] = [] - if dev_ip: - ips.append( - Interface.IP(ip_str=dev_ip, default_gateway=_gateway_or_none(gw)) - ) - - iface_fields: Dict[str, Any] = { - "name": iface_name, - "mode": mode, - "network": network_fields, - "ips": ips, - } - - dev_rec["interfaces"].append(iface_fields) - - # 6) Instantiate Network objects - networks: Dict[str, Network] = {} - for n_name, fields in network_agg.items(): - # Network.__init__ у вас принимает dict-like "fields" - networks[n_name] = Network(fields) - - # 7) Instantiate Device objects (Host/Switch/Router) - devices: Dict[str, Device] = {} - for d_name, fields in devices_raw.items(): - role = _clean(fields.get("role", "")).lower() - cls = ROLE_MAP.get( - role, Device - ) # если роль неизвестна — создадим базовый Device - devices[d_name] = cls(fields) - - return ParseResult( - devices_raw=devices_raw, - devices=devices, - networks=networks, - network_interfaces=network_interfaces, - ) - - -if __name__ == "__main__": - result = parse_csv_topology("table.csv") - - print("=== devices_raw (dict for debugging/serialization) ===") - for k, v in result.devices_raw.items(): - print(k, "=>", v) - - print("\n=== networks (Network objects) ===") - for k, v in result.networks.items(): - print(k, "=>", v) - - print("\n=== network_interfaces (Network -> [(device, iface), ...]) ===") - for net, items in result.network_interfaces.items(): - print(net, "=>", items) - - print("\n=== devices (Device objects) ===") - for k, v in result.devices.items(): - print(k, "=>", v) diff --git a/src_old/structures.py b/src_old/structures.py deleted file mode 100644 index dea9981..0000000 --- a/src_old/structures.py +++ /dev/null @@ -1,134 +0,0 @@ -from dataclasses import dataclass, field -from typing import List, Optional, Dict - -# =========================== -# Network / Subnet -# =========================== - -@dataclass -class Network: - name: str - vlan_id : Optional[int] = None - ip: Optional[str] = None - subnet_mask: Optional[str] = None - - def __init__(self, fields): - self.name = fields.get("name") - self.vlan_id = fields.get("vlan", fields.get("vlan_id")) - self.ip = fields.get("ip") - self.subnet_mask = fields.get("mask", fields.get("subnet_mask")) - self.__post_init__() - - def __post_init__(self): - if not isinstance(self.name, str) or not self.name: - raise ValueError("Network 'name' must be a non-empty string") - -# =========================== -# Interface Classes -# =========================== - - -@dataclass -class Interface: - class IP: - def __init__(self, ip_str: str, default_gateway: Optional[str] = None): - self.ip_str = ip_str - self.default_gateway = default_gateway - - def __str__(self): - return self.ip_str - - name: str - mode: Optional[str] = None - network: Optional[Network] = None - ips: Optional[List[IP]] = None - mac_address: Optional[str] = None - speed: Optional[str] = None - - def __init__(self, fields): - self.name = fields.get("name") - ips = fields.get("ips", []) - if isinstance(ips, str): - ips = [self.IP(ip_str=ip.strip()) for ip in ips.split(",") if ip.strip()] - self.ips = ips - - # возможно передаём просто field - self.network = Network(fields.get("network", {})) if fields.get("network") else None - - self.mac_address = fields.get("mac") - self.speed = fields.get("speed") - self.__post_init__() - - def __post_init__(self): - if not isinstance(self.name, str) or not self.name: - raise ValueError("Interface 'name' must be a non-empty string") - - -# =========================== -# Device Base Class (Abstract) -# =========================== - - -@dataclass -class Device: - name: str - interfaces: List[Interface] = field(default_factory=list) - - def __init__(self, fields): - self.name = fields.get("name") - # тут точно понять, какие поля передаются - self.interfaces = [Interface(iface) for iface in fields.get("interfaces", [])] - self.__post_init__() - - def __post_init__(self): - if not isinstance(self.name, str) or not self.name: - raise ValueError("Device 'name' must be a non-empty string") - - -# =========================== -# Host / Router / Switch -# =========================== - - -@dataclass -class Host(Device): - # operating_system: Optional[str] = None - - def __init__(self, fields): - super().__init__(fields) - # self.operating_system = fields.get("operatingSystem") - self.__post_init__() - - def __post_init__(self): - super().__post_init__() - - -@dataclass -class Router(Device): - # routing_table: Dict[str, str] = field(default_factory=dict) - # routing_protocols: List[str] = field(default_factory=list) - - def __init__(self, fields): - super().__init__(fields) - # self.routing_table = fields.get("routingTable", {}) - # self.routing_protocols = fields.get("routingProtocols", []) - self.__post_init__() - - def __post_init__(self): - super().__post_init__() - - -@dataclass -class Switch(Device): - # mac_table: Dict[str, str] = field(default_factory=dict) - # vlan_database: Dict[int, str] = field(default_factory=dict) - - def __init__(self, fields): - super().__init__(fields) - # self.mac_table = fields.get("macTable", {}) - # self.vlan_database = fields.get("vlanDatabase", {}) - self.__post_init__() - - def __post_init__(self): - super().__post_init__() -