导入必要的库

import paramiko from ncclient import manager import time import datetime

定义交换机类

class Switch: def init(self, ip, username, password, port=22): self.ip = ip self.username = username self.password = password self.port = port self.ssh = None self.connect()

# 连接交换机
def connect(self):
    self.ssh = paramiko.SSHClient()
    self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    self.ssh.connect(self.ip, self.port, self.username, self.password)

# 执行命令
def execute_command(self, command):
    stdin, stdout, stderr = self.ssh.exec_command(command)
    output = stdout.read().decode()
    return output

# 关闭连接
def close(self):
    if self.ssh:
        self.ssh.close()

# 获取电源状态
def get_power_status(self):
    command = 'display power'
    output = self.execute_command(command)
    if 'off' in output:
        return 'faulty'
    else:
        return 'normal'

# 获取风扇状态
def get_fan_status(self):
    command = 'display fan'
    output = self.execute_command(command)
    if 'faulty' in output:
        return 'faulty'
    else:
        return 'normal'

# 获取LACP状态
def get_lacp_status(self):
    command = 'display lacp'
    output = self.execute_command(command)
    if 'down' in output:
        return 'faulty'
    else:
        return 'normal'

# 获取CPU利用率
def get_cpu_utilization(self):
    command = 'display cpu-usage'
    output = self.execute_command(command)
    utilization = output.split('

')[2].split()[0].strip('%') if float(utilization) >= 80: return 'faulty' else: return 'normal'

# 获取内存利用率
def get_memory_utilization(self):
    command = 'display memory-usage'
    output = self.execute_command(command)
    utilization = output.split('

')[2].split()[0].strip('%') if float(utilization) >= 80: return 'faulty' else: return 'normal'

# 获取OSPF邻居状态
def get_ospf_neighbor_status(self):
    command = 'display ospf neighbor'
    output = self.execute_command(command)
    if 'Full' not in output:
        return 'faulty'
    else:
        return 'normal'

# 监控关键运行状态
def monitor_status(self):
    status = {
        'power': self.get_power_status(),
        'fan': self.get_fan_status(),
        'lacp': self.get_lacp_status(),
        'cpu': self.get_cpu_utilization(),
        'memory': self.get_memory_utilization(),
        'ospf': self.get_ospf_neighbor_status()
    }
    return status

定义配置类

class Config: def init(self, ip, username, password, port=830): self.ip = ip self.username = username self.password = password self.port = port self.mgr = None self.connect()

# 连接设备
def connect(self):
    self.mgr = manager.connect(host=self.ip,
                               port=self.port,
                               username=self.username,
                               password=self.password,
                               hostkey_verify=False)

# 配置日志主机
def set_log_host(self):
    config = '''
    <config>
        <system xmlns='urn:huawei:yang:huawei-system'>
            <log>
                <host>
                    <type>server</type>
                    <host-ip>10.1.60.2</host-ip>
                </host>
            </log>
        </system>
    </config>
    '''
    self.mgr.edit_config(target='running', config=config)

# 关闭连接
def close(self):
    if self.mgr:
        self.mgr.close_session()

读取监控命令

def read_commands(): with open('commands.txt') as f: commands = f.read().splitlines() return commands

监控交换机状态

def monitor_switch(switch): # 获取当前时间 now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 获取状态 status = switch.monitor_status() # 输出状态 print(f'{now}: Power status: {status['power']}, Fan status: {status['fan']}, LACP status: {status['lacp']}, CPU utilization: {status['cpu']}, Memory utilization: {status['memory']}, OSPF neighbor status: {status['ospf']}') # 分析状态，如果有异常则给出提示 if status['fan'] == 'faulty': print(f'{now}: Fan is faulty!') if status['memory'] == 'faulty': print(f'{now}: Memory utilization is high!') # 每24h保存一次配置文件并备份 if now.endswith('00:00:00'): config = switch.execute_command('display current-configuration') with open(f'{switch.ip}_config.txt', 'w') as f: f.write(config) sftp = paramiko.Transport(('10.1.60.2', 22)) sftp.connect(username='username', password='password') sftp.put(f'{switch.ip}_config.txt', f'{switch.ip}_config.txt') sftp.close()

主函数

def main(): # 初始化交换机对象 switch = Switch(ip='10.1.1.1', username='username', password='password') # 初始化配置对象 config = Config(ip='10.1.1.1', username='username', password='password') # 打开NETCONF功能 config.set_log_host() # 读取监控命令 commands = read_commands() # 循环监控交换机状态 while True: for command in commands: switch.execute_command(command) monitor_switch(switch) time.sleep(300) # 每5分钟监控一次

# 关闭连接
switch.close()
config.close()

if name == 'main': main()