import re import sys from collections import defaultdict from pathlib import Path from typing import List, Optional, Tuple from dataclasses import dataclass from scripts.format import FormatMain @dataclass class NetworkDevice: """Classe de base pour représenter un équipement réseau.""" name: str macs: List[str] interface: str dest_mac: str bridge: str vitesse: str nb_liens: str def __post_init__(self): """Validation et nettoyage des données après initialisation.""" if isinstance(self.macs, str): self.macs = [mac.strip() for mac in self.macs.split(',')] self.bridge = self.bridge.strip() self.vitesse = self.vitesse.strip() self.nb_liens = self.nb_liens.strip() @property def macs_str(self) -> str: """Retourne les MACs sous forme de chaîne formatée.""" return ','.join(self.macs) def to_formatted_string(self) -> str: """Retourne une représentation formatée de l'équipement.""" return f"{self.name} [{self.macs_str}] {self.interface} -> {self.dest_mac} [{self.bridge},{self.vitesse},{self.nb_liens}]" @dataclass class Coeur(NetworkDevice): """Classe représentant un équipement coeur.""" pass @dataclass class Switch(NetworkDevice): """Classe représentant un switch.""" pass class NetworkFileParser: """Classe pour parser les fichiers de configuration réseau.""" ENCODINGS = ['utf-8', 'utf-16', 'utf-16-le', 'utf-16-be', 'latin-1', 'cp1252', 'iso-8859-1'] PATTERNS = [ r'(\S+)\s+\[([^\]]+)\]\s+(\S+)\s+->\s+([A-Fa-f0-9:-]+)\s*\[([^,]+),([^,]+),([^\]]+)\]', r'(\S+)\s+\[([^\]]+)\]\s+(\S+)\s*->\s*([A-Fa-f0-9:-]+)\s*\[([^,]+),([^,]+),([^\]]+)\]', r'(\w+)\s*\[([^\]]+)\]\s*(\w+)\s*->\s*([A-Fa-f0-9:-]+)\s*\[([^,]+),([^,]+),([^\]]+)\]' ] def __init__(self, file_path: str): self.file_path = Path(file_path) def _read_file_with_encoding(self) -> Optional[str]: """Lit le fichier en essayant différents encodages.""" for encoding in self.ENCODINGS: try: return self.file_path.read_text(encoding=encoding) except (UnicodeDecodeError, UnicodeError): continue except FileNotFoundError: return None return None def _parse_line(self, line: str, device_class) -> Optional[NetworkDevice]: """Parse une ligne et retourne un objet NetworkDevice.""" line = line.strip() if not line: return None for pattern in self.PATTERNS: match = re.match(pattern, line) if match: return device_class( name=match.group(1), macs=match.group(2), interface=match.group(3), dest_mac=match.group(4), bridge=match.group(5), vitesse=match.group(6), nb_liens=match.group(7) ) print(f" -> Aucun pattern ne correspond à cette ligne: {line}") return None def parse(self) -> Tuple[List[Coeur], List[Switch]]: """Parse le fichier et retourne les coeurs et switches.""" content = self._read_file_with_encoding() if content is None: return [], [] sections = content.split('Switches:') if len(sections) < 2: return [], [] coeur_section = sections[0].replace('Coeur:', '').strip() switches_section = sections[1].strip() coeurs = [] for line in coeur_section.split('\n'): coeur = self._parse_line(line, Coeur) if coeur: coeurs.append(coeur) switches = [] for line in switches_section.split('\n'): switch = self._parse_line(line, Switch) if switch: switches.append(switch) return coeurs, switches class NetworkDeviceGrouper: """Classe pour grouper les équipements réseau.""" @staticmethod def group_coeurs_by_dest_mac(coeurs: List[Coeur]) -> List[Coeur]: """Groupe les coeurs par adresse MAC de destination.""" grouped = defaultdict(list) for coeur in coeurs: key = (coeur.dest_mac, coeur.name, coeur.macs_str, coeur.bridge, coeur.vitesse, coeur.nb_liens) grouped[key].append(coeur) result = [] for key, group in grouped.items(): dest_mac, name, macs_str, bridge, vitesse, nb_liens = key interfaces = [item.interface for item in group] result.append(Coeur( name=name, macs=macs_str.split(','), interface='-'.join(interfaces), dest_mac=dest_mac, bridge=bridge, vitesse=vitesse, nb_liens=nb_liens )) return result @staticmethod def group_switches_by_local_mac(switches: List[Switch]) -> List[Switch]: """Groupe les switches par adresse MAC locale.""" grouped = defaultdict(list) for switch in switches: key = (switch.name, switch.macs_str, switch.dest_mac, switch.bridge, switch.vitesse, switch.nb_liens) grouped[key].append(switch) result = [] for key, group in grouped.items(): name, macs_str, dest_mac, bridge, vitesse, nb_liens = key interfaces = [item.interface for item in group] result.append(Switch( name=name, macs=macs_str.split(','), interface='-'.join(interfaces), dest_mac=dest_mac, bridge=bridge, vitesse=vitesse, nb_liens=nb_liens )) return result class MermaidDiagramGenerator: """Classe pour générer les diagrammes Mermaid.""" def __init__(self, coeurs: List[Coeur], switches: List[Switch]): self.coeurs = coeurs self.switches = switches def _find_matching_switch(self, coeur: Coeur) -> Optional[Switch]: """Trouve le switch correspondant à un coeur.""" for switch in self.switches: if coeur.dest_mac in switch.macs: return switch return None def _format_bridge_label(self, bridge: str, vitesse: str) -> str: """Formate le label du bridge.""" bridge_formatted = bridge.replace('Bridge-Aggregation', 'BAGG') return f"{bridge_formatted}
{vitesse}" def generate_links_diagram(self) -> str: """Génère un diagramme Mermaid des liaisons.""" if not self.coeurs or not self.switches: return "Aucune donnée à afficher" mermaid_lines = ["graph LR"] total_devices = min(len(self.coeurs), len(self.switches)) for i, coeur in enumerate(self.coeurs): target_switch = self._find_matching_switch(coeur) label = self._format_bridge_label(coeur.bridge, coeur.vitesse) if target_switch: if i < total_devices // 2: line = f' Coeur(("{coeur.name}")) <-->|{label}| {target_switch.name}' else: line = f' {target_switch.name} <-->|{label}| Coeur(("{coeur.name}"))' mermaid_lines.append(line) else: line = f' Coeur(("{coeur.name}")) <-->|{label}| {coeur.dest_mac}["📄manquant {coeur.dest_mac}"]' mermaid_lines.append(line) mermaid_lines.append(f' class {coeur.dest_mac} error;') return '\n'.join(mermaid_lines) class FileWriter: """Classe pour l'écriture de fichiers.""" @staticmethod def write_mermaid_diagram(mermaid_code: str, output_path: Path) -> None: """Sauvegarde le diagramme Mermaid.""" try: output_path.parent.mkdir(parents=True, exist_ok=True) content = [ "# Diagramme des liaisons Coeur-Switch\n", "```mermaid\n", "---\n", "config:\n", " theme: 'base'\n", " themeVariables:\n", " primaryColor: '#25bb75ff'\n", " primaryTextColor: '#ffffff'\n", " primaryBorderColor: '#000000ff'\n", " lineColor: '#f82929ff'\n", " secondaryColor: '#5e0c1aff'\n", " tertiaryColor: '#ffffff'\n", "---\n", mermaid_code, "\nclass Coeur coeur;\n", "classDef coeur fill:#0590e6;\n", "classDef error fill:#e64c05;\n", "\n```\n" ] output_path.write_text(''.join(content), encoding='utf-8') print(f"✅ Diagramme Mermaid généré : {output_path}") except Exception as e: print(f"Erreur lors de la sauvegarde: {e}") @staticmethod def write_grouped_file(coeurs: List[Coeur], switches: List[Switch], output_path: Path) -> None: """Écrit les équipements groupés dans un fichier.""" try: lines = ["Coeur:\n"] lines.extend(f"{coeur.to_formatted_string()}\n" for coeur in coeurs) lines.append("Switches:\n") lines.extend(f"{switch.to_formatted_string()}\n" for switch in switches) output_path.write_text(''.join(lines), encoding='utf-8') except Exception as e: print(f"Erreur lors de la sauvegarde: {e}") @staticmethod def write_links_file(coeurs: List[Coeur], switches: List[Switch], output_path: Path) -> None: """Écrit les liens entre coeurs et switches.""" try: lines = [] for coeur in coeurs: matching_switch = None for switch in switches: if coeur.dest_mac in switch.macs: matching_switch = switch break if matching_switch: coeur_info = f"{coeur.name} [{coeur.macs_str}] {coeur.interface} [{coeur.bridge},{coeur.vitesse},{coeur.nb_liens}]" switch_info = f"{matching_switch.name} [{matching_switch.macs_str}] {matching_switch.interface} [{matching_switch.bridge},{matching_switch.vitesse},{matching_switch.nb_liens}]" lines.append(f"{coeur_info} -> {switch_info}\n") else: lines.append(f"{coeur.name} [{coeur.macs_str}] {coeur.interface} -> Aucune correspondance de switch pour MAC {coeur.dest_mac}\n") output_path.write_text(''.join(lines), encoding='utf-8') except Exception as e: print(f"Erreur lors de l'écriture des liens: {e}") class NetworkAnalyzer: """Classe principale pour l'analyse réseau.""" def __init__(self, base_dir: Path = None): if base_dir is None: base_dir = Path(__file__).parent.parent self.base_dir = Path(base_dir) self.data_file = self.base_dir / 'data.txt' self.output_dir = self.base_dir / 'output' self.mermaid_file = self.output_dir / 'mermaid.md' def analyze(self, filename: str) -> None: """Analyse complète du réseau.""" # Exécute l'analyse initiale FormatMain().run_analysis(filename) # Parse le fichier parser = NetworkFileParser(self.data_file) coeurs, switches = parser.parse() if not coeurs and not switches: print("Impossible de lire le fichier ou format incorrect") return # Groupe les équipements grouper = NetworkDeviceGrouper() grouped_coeurs = grouper.group_coeurs_by_dest_mac(coeurs) grouped_switches = grouper.group_switches_by_local_mac(switches) # Écrit les fichiers de sortie writer = FileWriter() writer.write_grouped_file(grouped_coeurs, grouped_switches, self.data_file) writer.write_links_file(grouped_coeurs, grouped_switches, self.data_file) # Génère le diagramme Mermaid diagram_generator = MermaidDiagramGenerator(grouped_coeurs, grouped_switches) mermaid_code = diagram_generator.generate_links_diagram() writer.write_mermaid_diagram(mermaid_code, self.mermaid_file) def main(): """Fonction principale.""" if len(sys.argv) != 2: print("Usage: python mermaid.py ") sys.exit(1) analyzer = NetworkAnalyzer() analyzer.analyze(filename=sys.argv[1]) if __name__ == "__main__": main()