diff --git a/1НФ VLAN - 1НФ.csv b/1НФ VLAN - 1НФ.csv new file mode 100644 index 0000000..f3b8583 --- /dev/null +++ b/1НФ VLAN - 1НФ.csv @@ -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,,,, \ No newline at end of file diff --git a/output.yaml b/output.yaml index 220ed9a..45278eb 100644 --- a/output.yaml +++ b/output.yaml @@ -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 diff --git a/src/make_res.py b/src/make_res.py index ee4f06d..1ce5ec9 100644 --- a/src/make_res.py +++ b/src/make_res.py @@ -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 -""" -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" -""" +def _nonempty(v: Any) -> bool: + if v is None: + return False + if isinstance(v, str) and not v.strip(): + return False + return True -def make_res(data, name): - res = { - "meta": { - "id": name, - "name": name, - }, +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": [], } - host_int = {} + # ----------------- + # 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] + ] - 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) + net_entry: Dict[str, Any] = {"members": members} - host_int[host.interfaces[0]] = host.name + 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 - 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]], - ] - } - res["links"].append(link_entry) + # vlan (int) + if _nonempty(net_obj.vlan_id): + net_entry["vlan"] = net_obj.vlan_id - return res + # 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__": - 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() diff --git a/src/parse.py b/src/parse.py index ea579a7..6cf65ec 100644 --- a/src/parse.py +++ b/src/parse.py @@ -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: - continue - result[obj] = {} - for header, value in zip(headers, row[1:]): - value = value.strip() +def _to_int(v: str) -> Optional[int]: + v = _clean(v) + if v.isdigit(): + return int(v) + return None - if value: - result[obj][header] = value - return result +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__": - 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) diff --git a/table.csv b/table.csv new file mode 100644 index 0000000..f3b8583 --- /dev/null +++ b/table.csv @@ -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,,,, \ No newline at end of file diff --git a/table.yaml b/table.yaml new file mode 100644 index 0000000..8eababb --- /dev/null +++ b/table.yaml @@ -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