utf-8 python3

导入模块

from ncclient import manager from paramiko import SSHClient, AutoAddPolicy import datetime import time

设备信息变量

div_ip = '10.1.0.6' ssh_usr = 'python' ssh_pwd = 'Huawei@123'

cn_usr = 'ncclient' cn_pwd = 'Huawei@123'

监控命令保存文件

Command_file='command.txt'

保存配置

save= datetime.datetime.now().strftime('%Y-%m-%d-%H')+ '_X_T1_AGG1.zip'

设备类

class Device(): def init(self,ip,usr,pwd): self.host =ip # 设备ip self.username=usr# 设备用户 self.password=pwd# 设备密码 self.client = self.get_connect() # 调用 SSH 连接函数 self.cli = self.client.invoke_shell() # 使用 shell 交互 self.cli.send('n\n') # 操作步骤 self.cli.send('\n') # 分页设置 self.cli.recv(65535)

# SSH 连接函数
def get_connect(self):
    sshcon = SSHClient()
    sshcon.set_missing_host_key_policy(AutoAddPolicy)  # 未知主机自动添加密钥
    sshcon.connect(hostname=self.host,username=self.username,password=self.password)  # 连接信息
    return sshcon

# 监控设备以及输出
def cmd_running(self):
    file = open(Command_file, 'r') # 打开监控文件并读取
    cmd_list = file.readlines()  # 以字符串的形式读取
    for cmd in cmd_list: # 历遍所有命令
        self.cli.send(cmd)
        time.sleep(1)
        dis_th = self.cli.recv(65535).decode('utf-8') # 以文本形式输出
        if cmd == 'display fan\n':  # 输入的命令等于 display fan
            status = dis_th.find('normal') # 查找 normal 状态
            if status < 0 : # 如果状态不存在输出下面结果
                print('All fans are faulty.')
        else:   # 否则输出下面结果
            print(dis_th)
        # 输出到文本
        time_n= datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S') # 生成时间戳
        file_name=f'{self.host}_{time_n}.txt'
        with open(file_name, 'w') as f:
            f.write(dis_th)


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

# 保存配置
def save_running(self):
    cmd = 'save ' + save # 保存命令
    self.cli.send(cmd) # 发送保存命令
    self.cli.send('y\n') # 确认保存
    self.cli.send('y\n') # 已存在, 覆盖确认保存
    time.sleep(1)
    dis_th=self.cli.recv(65535).decode('utf-8')  # 以文本形式输出
    print(dis_th)

# 下载保存配置
def download(self):
    ssh_con = self.get_connect() # 使用 ssh 连接
    sftp_con = ssh_con.open_sftp() # 打开 sftp
    l_path = save   # 本地路径
    r_path = '/' +l_path # 远端路径
    sftp_con.get(remoutepath=r_path,localpath=l_path) # 使用 get 方式下载
    ssh_con.close() # 关闭 ssh 连接

# 打开 netconf 功能
def netconf_run(self):
    self.cli.send('sys\n') # 发送命令并换行
    time.sleep(1) # 等待进入配置
    self.cli.send('netconf\n') # 发送 netconf 命令
    time.sleep(10)  # 等待 netconf 功能开启
    self.cli.send('y\n') # 如已经开启, 确认在其配置
    self.cli.send('protocol inbound ssh port 830\n') # 配置端口
    time.sleep(1)
    dis_th=self.cli.recv(65535).decode('utf-8')  # 以文本形式输出
    print(dis_th)

调用函数

def netconf(ip,usr,pwd): ''' 启用 netconf 功能 ''' netconf = Device(ip,usr,pwd) netconf.netconf_run() netconf.close()

循环保存

def monitor(ip,usr,pwd): ''' 监控设备并保存配置 ''' while True: monitor = Device(ip,usr,pwd) monitor.cmd_running() now =datetime.datetime.now() save_time_24h=now.replace(hour=0,minute=0,second=0,microsecond=0)+datetime.timedelta(days=1) save_time_5min=now +datetime.timedelta(minutes=5) if now >= save_time_24h: monitor.save_running() monitor.download() elif now >= save_time_5min: monitor.save_running() monitor.download() time.sleep(300)

if name == 'main': try: netconf(div_ip,ssh_usr,ssh_pwd) monitor(div_ip,ssh_usr,ssh_pwd) except KeyboardInterrupt : print('已经停止')