update src

This commit is contained in:
2025-12-22 18:50:23 +03:00
parent c4856cdcb6
commit 612c2ad6b2
6 changed files with 551 additions and 95 deletions
+11
View File
@@ -0,0 +1,11 @@
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
+14 -11
View File
@@ -1,16 +1,19 @@
links:
- endpoints:
- PC1
- PC2
networks:
- A: [PC1.eth0, PC2.eth0]
meta:
id: /home/krosh/Documents/Github/network-diagrams-tool/host-host.csv
name: /home/krosh/Documents/Github/network-diagrams-tool/host-host.csv
id: host-host.csv
name: host-host.csv
nodes:
- interfaces:
- ip: 10.0.12.1/24
- role: host
name: PC1
role: host
- interfaces:
- ip: 10.0.12.2/24
interfaces:
- eth0:
- ip: 10.0.12.1/24
network: A
- role: host
name: PC2
role: host
interfaces:
- eth0:
- ip: 10.0.12.2/24
network: A
+236 -66
View File
@@ -1,81 +1,251 @@
from check_correct import check_correct
from parse import parse_csv_to_structure
import sys
from structures import Host, Wire, Interface
import yaml
# 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:
"""
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"
Маска в 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 make_res(data, name):
res = {
"meta": {
"id": name,
"name": name,
},
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": [],
}
host_int = {}
for key, value in data.items():
if value.get("type") == "host":
value["name"] = key
host = Host(value)
node_entry = {
"name": host.name,
"role": "host",
"interfaces": [
{"ip": Interface(data[iface]).ip_address}
for iface in host.interfaces
],
}
res["nodes"].append(node_entry)
host_int[host.interfaces[0]] = host.name
for key, value in data.items():
if value.get("type") == "wire":
wire = Wire(value)
link_entry = {
"endpoints": [
host_int[wire.endpoints[0]],
host_int[wire.endpoints[1]],
# -----------------
# 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]
]
}
res["links"].append(link_entry)
return res
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__":
data = parse_csv_to_structure(sys.argv[1])
print(data)
if check_correct(data):
print("Data is correct.")
res = make_res(data, sys.argv[1])
print(res)
with open("output.yaml", "w") as f:
yaml.dump(res, f)
main()
+197 -17
View File
@@ -1,31 +1,211 @@
# 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
def parse_csv_to_structure(path):
result = {}
# ===========================
# 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",
}
with open(path, newline="") as f:
reader = csv.reader(f)
rows = list(reader)
ROLE_MAP = {
"host": Host,
"switch": Switch,
"router": Router,
}
headers = rows[0][1:]
for row in rows[1:]:
obj = row[0].strip()
def _clean(v: Any) -> str:
return str(v).strip() if v is not None else ""
if not obj:
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
result[obj] = {}
for header, value in zip(headers, row[1:]):
value = value.strip()
# 2) init device record
dev_rec = devices_raw.setdefault(
device_name,
{"name": device_name, "role": role, "interfaces": []},
)
if value:
result[obj][header] = value
# если роль у устройства не была проставлена ранее, но появилась сейчас — заполним
if role and not dev_rec.get("role"):
dev_rec["role"] = role
return result
# если интерфейс не указан — дальше нечего собирать
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__":
data = parse_csv_to_structure("input.csv")
print(data)
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)
+11
View File
@@ -0,0 +1,11 @@
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
+81
View File
@@ -0,0 +1,81 @@
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