krosh/network-diagrams-tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

remove old files

Browse Source
This commit is contained in:
krosh
2026-02-24 21:53:04 +03:00
parent 1ae7a5e78a
commit cefef205df
16 changed files with 1 additions and 1014 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1 -1
View File
@@ -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/
.vscode/settings.json
Vendored
-3
View File
@@ -1,3 +0,0 @@
{
"nixEnvSelector.nixFile": "${workspaceFolder}/flake.nix"
}
UML/structure.jpg
BIN
View File
Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 186 KiB

data/input/1НФ VLAN - 1НФ.csv
-11
View File
@@ -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,,,,
1 Name Role Interface Network VLAN Network IP Mask Device IP Default Gateway
2 PC1 Host eth1 A 4 10.0.0.0 /24 10.0.0.1 0.0.0.0
3 PC2 Host eth1 B 2 10.0.0.0 /24 10.0.0.2 0.0.0.0
4 PC3 Host eth1 C 3 10.0.0.0 /24 10.0.0.3 0.0.0.0
5 PC4 Host eth1 D 4 10.0.0.0 /24 10.0.0.4 0.0.0.0
6 com_left Switch eth1 tr trunk
7 com_left Switch eth2 A 4
8 com_left Switch eth3 B 2
9 com_right Switch eth1 tr trunk
10 com_right Switch eth2 D 4
11 com_right Switch eth3 C 3
data/input/host-host.csv
-6
View File
@@ -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"
1 name type interfaces ip endpoints
2 PC1 host int1
3 PC2 host int2
4 int1 interface 10.0.12.1/24
5 int2 interface 10.0.12.2/24
6 link wire int1,int2
data/input/table.csv
-11
View File
@@ -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,,,,
1 Name Role Interface Network VLAN Network IP Mask Device IP Default Gateway
2 PC1 Host eth1 A 4 10.0.0.0 /24 10.0.0.1 0.0.0.0
3 PC2 Host eth1 B 2 10.0.0.0 /24 10.0.0.2 0.0.0.0
4 PC3 Host eth1 C 3 10.0.0.0 /24 10.0.0.3 0.0.0.0
5 PC4 Host eth1 D 4 10.0.0.0 /24 10.0.0.4 0.0.0.0
6 com_left Switch eth1 tr trunk
7 com_left Switch eth2 A 4
8 com_left Switch eth3 B 2
9 com_right Switch eth1 tr trunk
10 com_right Switch eth2 D 4
11 com_right Switch eth3 C 3
data/output/diagram.png
BIN
View File
Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 22 KiB

data/output/output.yaml
-19
View File
@@ -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
data/output/schema.yaml
-21
View File
@@ -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"
data/output/table.yaml
-81
View File
@@ -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
src_old/check_correct.py
-22
View File
@@ -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)
src_old/make_res.py
-251
View File
@@ -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: <csv>.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()
src_old/old/structures copy.py
-61
View File
@@ -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
src_old/old/structures.py
-182
View File
@@ -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")
src_old/parse.py
-211
View File
@@ -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)
src_old/structures.py
-134
View File
@@ -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__()