导入需要的模块

import paramiko import ncclient import time import datetime

定义 HuaweiDevice 类

class HuaweiDevice: # 初始化方法，传入设备 IP 地址、用户名和密码 def init(self, ip, username, password): self.ip = ip self.username = username self.password = password self.ssh = None self.netconf = None

# SSH 连接方法
def connect_ssh(self):
    self.ssh = paramiko.SSHClient()
    self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    self.ssh.connect(self.ip, username=self.username, password=self.password)

# SSH 断开连接方法
def disconnect_ssh(self):
    if self.ssh:
        self.ssh.close()
        self.ssh = None

# 执行 SSH 命令方法，传入命令字符串
def execute_ssh_command(self, command):
    if not self.ssh:
        self.connect_ssh()
    stdin, stdout, stderr = self.ssh.exec_command(command)
    return stdout.readlines()

# NETCONF 连接方法
def connect_netconf(self):
    self.netconf = ncclient.manager.connect(
        host=self.ip,
        port=830,
        username=self.username,
        password=self.password,
        hostkey_verify=False
    )

# NETCONF 断开连接方法
def disconnect_netconf(self):
    if self.netconf:
        self.netconf.close_session()
        self.netconf = None

# 执行 NETCONF 命令方法，传入命令字符串
def execute_netconf_command(self, command):
    if not self.netconf:
        self.connect_netconf()
    return self.netconf.edit_config(target='running', config=command)

# 监控设备状态方法，传入设备名称
def monitor_device(self, device_name):
    while True:
        # # 电源状态
        # power_status = self.execute_ssh_command(f'display device {device_name} power')
        # print('Power status:')
        # print(power_status)
        #
        # # 风扇状态
        # fan_status = self.execute_ssh_command(f'display device {device_name} fan')
        # print('Fan status:')
        # print(fan_status)

        # LACP 状态
        lacp_status = self.execute_ssh_command(f'display lacp statistics eth-trunk')
        print('LACP status:')
        print(lacp_status)

        # CPU、内存利用率
        cpu_usage = self.execute_ssh_command(f'display cpu-usage')
        print('CPU usage:')
        print(cpu_usage)
        memory_usage = self.execute_ssh_command(f'display memory-usage')
        print('Memory usage:')
        print(memory_usage)

        # OSPF 邻居状态
        ospf_status = self.execute_ssh_command('display ospf peer')
        print('OSPF status:')
        print(ospf_status)

        # # 判断风扇状态是否异常
        # fan_status = [line.strip() for line in fan_status if line.strip()]
        # if len(fan_status) >= 2 and all('Normal' not in line for line in fan_status):
        #     print('All fans are faulty!')

        time.sleep(300)  # 等待 5 分钟

# 备份配置方法，传入设备名称
def backup_config(self, device_name):
    # 获取当前时间
    now = datetime.datetime.now().strftime('%Y_%m_%d')
    # 备份文件名
    backup_filename = f'{now}_{device_name}'
    # 压缩文件名
    config_filename = f'{backup_filename}.zip'
    # 备份命令字符串
    backup_command = f'save configuration to {backup_filename}.cfg\n'
    backup_command += f'file compress {backup_filename}.cfg {config_filename}\n'
    self.execute_netconf_command(backup_command)

    # 通过 SFTP 下载文件到本地
    transport = paramiko.Transport((self.ip, 22))
    transport.connect(username=self.username, password=self.password)
    sftp = paramiko.SFTPClient.from_transport(transport)
    sftp.get(config_filename, config_filename)
    sftp.close()
    transport.close()

# 开启 NETCONF 方法
def enable_netconf(self):
    command = '''
        <config>
            <netconf xmlns='urn:ietf:params:xml:ns:netconf:base:1.0'>
                <enabled>true</enabled>
            </netconf>
        </config>
    '''
    self.execute_netconf_command(command)

# 配置日志服务器方法，传入设备名称和日志服务器 IP 地址
def configure_log_server(self, device_name, log_server_ip):
    command = f'''
        <config>
            <logging xmlns='http://www.huawei.com/netconf/vrp' content-version='1.0' format-version='1.0'>
                <host>
                    <name>{log_server_ip}</name>
                    <device-device-id>{device_name}</device-device-id>
                </host>
            </logging>
        </config>
    '''
    self.execute_netconf_command(command)

if name == 'main': # 实例化 HuaweiDevice 类 device = HuaweiDevice(ip='192.168.1.1', username='admin', password='password') # 开启 NETCONF device.enable_netconf() # 配置日志服务器 device.configure_log_server(device_name='X_T1_ACC2', log_server_ip='10.1.60.2') # 监控设备状态 device.monitor_device(device_name='X_T1_ACC2') # 备份配置 device.backup_config(device_name='X_T1_ACC2')