from typing import Dict, List, Optional import logging from cisco_ssh_manager import cisco_ssh_manager logger = logging.getLogger(__name__) class UniversalDeviceManager: """Universal device manager for multiple vendors""" VENDOR_DEVICE_TYPES = { 'tplink': 'tplink_jetstream', 'ubiquiti': 'ubiquiti_edgeswitch', 'ubiquiti_edge': 'ubiquiti_edge', 'zte': 'zte_zxros', 'vsol': 'generic_termserver', 'fiberhome': 'generic_termserver', 'raisecom': 'generic_termserver', 'bdcom': 'generic_termserver', 'h3c': 'hp_comware', 'dlink': 'dell_os9', 'juniper': 'juniper', 'arista': 'arista_eos', 'fortinet': 'fortinet', 'paloalto': 'paloalto_panos', } def __init__(self): self.ssh_manager = cisco_ssh_manager def get_device_type(self, vendor: str) -> str: """Get netmiko device type for vendor""" vendor_lower = vendor.lower() return self.VENDOR_DEVICE_TYPES.get(vendor_lower, 'generic_termserver') async def execute_command(self, ip: str, username: str, password: str, command: str, vendor: str = 'generic', port: int = 22) -> str: """Execute command on any vendor device""" try: device_type = self.get_device_type(vendor) output = self.ssh_manager.execute_command( ip, username, password, command, device_type=device_type ) return output except Exception as e: logger.error(f"Failed to execute command on {vendor} device {ip}: {e}") raise # TP-Link Specific async def tplink_get_system_info(self, ip: str, username: str, password: str) -> Dict: """Get TP-Link system information""" try: output = await self.execute_command( ip, username, password, 'show system-info', vendor='tplink' ) return { 'raw_output': output, 'vendor': 'TP-Link' } except Exception as e: raise Exception(f"Failed to get TP-Link info: {e}") async def tplink_get_interfaces(self, ip: str, username: str, password: str) -> str: """Get TP-Link interface status""" return await self.execute_command( ip, username, password, 'show interface status', vendor='tplink' ) async def tplink_get_vlans(self, ip: str, username: str, password: str) -> str: """Get TP-Link VLAN configuration""" return await self.execute_command( ip, username, password, 'show vlan', vendor='tplink' ) # Ubiquiti Specific async def ubiquiti_get_system(self, ip: str, username: str, password: str) -> Dict: """Get Ubiquiti system information""" try: output = await self.execute_command( ip, username, password, 'show version', vendor='ubiquiti' ) return { 'raw_output': output, 'vendor': 'Ubiquiti' } except Exception as e: raise Exception(f"Failed to get Ubiquiti info: {e}") async def ubiquiti_get_interfaces(self, ip: str, username: str, password: str) -> str: """Get Ubiquiti interface status""" return await self.execute_command( ip, username, password, 'show interfaces', vendor='ubiquiti' ) async def ubiquiti_get_wireless(self, ip: str, username: str, password: str) -> str: """Get Ubiquiti wireless clients""" return await self.execute_command( ip, username, password, 'show wireless clients', vendor='ubiquiti' ) # ZTE Specific async def zte_get_version(self, ip: str, username: str, password: str) -> Dict: """Get ZTE device version""" try: output = await self.execute_command( ip, username, password, 'show version', vendor='zte' ) return { 'raw_output': output, 'vendor': 'ZTE' } except Exception as e: raise Exception(f"Failed to get ZTE info: {e}") async def zte_olt_get_onus(self, ip: str, username: str, password: str, pon_port: str = 'gpon-olt_1/1/1') -> str: """Get ZTE OLT ONUs""" return await self.execute_command( ip, username, password, f'show gpon onu state {pon_port}', vendor='zte' ) # VSOL Specific async def vsol_get_system(self, ip: str, username: str, password: str) -> Dict: """Get VSOL OLT system information""" try: output = await self.execute_command( ip, username, password, 'show system', vendor='vsol' ) return { 'raw_output': output, 'vendor': 'VSOL' } except Exception as e: raise Exception(f"Failed to get VSOL info: {e}") async def vsol_get_onus(self, ip: str, username: str, password: str, olt: int = 1, pon: int = 1) -> str: """Get VSOL ONUs""" return await self.execute_command( ip, username, password, f'show onu-info olt {olt} pon {pon}', vendor='vsol' ) # Juniper Specific async def juniper_get_chassis(self, ip: str, username: str, password: str) -> Dict: """Get Juniper chassis info""" try: output = await self.execute_command( ip, username, password, 'show chassis hardware', vendor='juniper' ) return { 'raw_output': output, 'vendor': 'Juniper' } except Exception as e: raise Exception(f"Failed to get Juniper info: {e}") async def juniper_get_interfaces(self, ip: str, username: str, password: str) -> str: """Get Juniper interface status""" return await self.execute_command( ip, username, password, 'show interfaces terse', vendor='juniper' ) # Fortinet Specific async def fortinet_get_system(self, ip: str, username: str, password: str) -> Dict: """Get Fortinet FortiGate system status""" try: output = await self.execute_command( ip, username, password, 'get system status', vendor='fortinet' ) return { 'raw_output': output, 'vendor': 'Fortinet' } except Exception as e: raise Exception(f"Failed to get Fortinet info: {e}") async def fortinet_get_interfaces(self, ip: str, username: str, password: str) -> str: """Get Fortinet interface status""" return await self.execute_command( ip, username, password, 'get system interface physical', vendor='fortinet' ) # Generic Commands async def get_running_config(self, ip: str, username: str, password: str, vendor: str = 'generic') -> str: """Get running configuration (vendor agnostic)""" commands = { 'cisco': 'show running-config', 'juniper': 'show configuration', 'huawei': 'display current-configuration', 'zte': 'show running-config', 'vsol': 'show running-config', 'tplink': 'show running-config', 'ubiquiti': 'show configuration', 'fortinet': 'show full-configuration', } command = commands.get(vendor.lower(), 'show running-config') return await self.execute_command( ip, username, password, command, vendor ) async def backup_config(self, ip: str, username: str, password: str, vendor: str = 'generic') -> str: """Backup device configuration""" return await self.get_running_config(ip, username, password, vendor) async def ping_test(self, ip: str, username: str, password: str, target: str, vendor: str = 'generic') -> str: """Execute ping test""" command = f'ping {target} count 5' return await self.execute_command(ip, username, password, command, vendor) # Global instance universal_device_manager = UniversalDeviceManager()