如何通过Python脚本批量采集华为交换机配置

实验目的

通过学习python网络运维自动化减少重复无意义的工作量，本次实验虽然只是一条命令，通过display current-configuration采集设备配置，但是在有大量设备需要采集配置文件时，又无相应的网管平台时，小而便捷的python成为了一种选择，而且可以python脚本可以根据需要自行增加需要的命令。

设备环境

通过ENSP模拟华为交换机，桥接云跟主机通信，SSH配置可达。

设备配置命令：

 

sysname SW1
# 不同设备IP不同，sysname 不同，其他配置一致
interface Vlanif1
 ip address 192.168.111.10 255.255.255.0
#
aaa
 local-user admin password cipher 

	 

	其余两台按这个模板配置


	桥接云配置

	完成设备配置后测试SSH连接是否正常


	通过CRT测试，三台设备都能正常登录

	python环境

	首先看一下项目的结构


	目录说明

	backup_Script包是本次实验的主要代码，分为了三个模块来写的。

	Dest目录是采集配置文件后保存的目录。

	log是日志文件保存目录。

	source是中存在一个entry_table.csv，这是填写登录IP跟用户名的。

	

	不能更改列数，更改IP跟用户名密码的时候，就按照这个来更改

	requirements.txt里面内容是项目的依赖包及其对应版本号的信息列表，即项目依赖关系清单，其作用是用来重新构建项目所需要的运行环境依赖。

	run.py作为本次实验的入口文件，在运行项目的时候就运行该文件

	环境搭建

	：版本Python 3.10.2

	环境的搭建就从创建虚拟环境开始写，python安装之类的参考官方网站


	在最初项目的结构是这样的，然后进入到text目录下构建虚拟环境。

	首先进入项目的目录，创建虚拟环境

	 
# ven是可变的
python -m venv ven


	 

	

	构建完成后在次查看目录，会发现目录中多了一个目录ven，这个ven就是所创建的虚拟环境。


	当然，虚拟环境创建好后不代表就结束了，还需要进入到虚拟环境中，称为激活虚拟环境

	 
# 通过该方法激活虚拟环境，激活后会有一个括号在前面
venScriptsactivate
# 以下是激活后的显示
(ven) D:	ext>
# 激活完成更新以下pip库
python.exe -m pip install --upgrade pip
# 安装所需依赖
pip install -r requirements.txt
# 通过入口文件启用
run.py


	 

	


	从控制台输出的日志可以看出，脚本已经成功运行结束，那么查看一下相关的目录是否存在这文件


	查看Dest目录，已经存在三个文件，打开看一下是否已经获取到内容


	查看日志文件夹

	

	也存在相关的日志

	python源码分享

	一、backup_Script包中的源码

	_init_.py

	 
from .Read_source import login
from .Logger import log
import paramiko
import time
import socket
import re
import sys

sys.path.append('backup_Script')
ssh_client = None
switch_with_authentication_issue = []
switch_not_reachable = []


def logon_failed():
    global switch_with_authentication_issue
    global switch_not_reachable
    if switch_with_authentication_issue and switch_not_reachable is None:
        log.error('无登录失败记录')
    else:
        log.error('登录失败，详细查看文件log/switch_not_reachable.txt为不可达，switch_with_authentication_issue.txt为认证失败')
        with open('./log/switch_with_authentication_issue.txt', 'w') as f:
            f.write('
'.join(switch_with_authentication_issue))
        with open('./log/switch_not_reachable.txt', 'w') as f:
            f.write('
'.join(switch_not_reachable))


def run():
    global ssh_client
    global switch_with_authentication_issue
    global switch_not_reachable
    ip_list = login.get_entry_ip()
    name_list = login.get_entry_name()
    passwd_list = login.get_entry_passwd()
    regex = 'sysname .*'
    for i in zip(ip_list[1:], name_list[1:], passwd_list[1:]):
        ip = i[0]
        name = i[1]
        passwd = i[2]
        try:
            ssh_client = paramiko.SSHClient()
            ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh_client.connect(
                hostname=ip,
                username=name,
                password=passwd,
                look_for_keys=False
            )
            print('连接成功：', ip)
            command = ssh_client.invoke_shell()
            command.send("screen-length 0 temporary
")
            command.send("display current-configuration
")
            time.sleep(2)
            output = command.recv(65535)
            print(output.decode('ascii'))
            save_file = re.search(regex, output.decode('ascii')).group()
            sw_name = save_file[8:].strip()
            f = open(f"./Dest/{sw_name}.txt", 'w')
            f.write(output.decode(encoding='UTF-8'))
            f.close()
        except paramiko.ssh_exception.AuthenticationException:
            log.critical(f"用户认证失败的{ip}.")
            switch_with_authentication_issue.append(ip)
        except socket.error:
            log.critical(f"{ip}不可达，请检查网络.")
            switch_not_reachable.append(ip)
    ssh_client.close()
    logon_failed()



	 

	Logger.py

	 
import logging
import os


class Logger:
    def __init__(self):
        selfdef_fmt = '%(asctime)s - %(funcName)s - %(levelname)s - %(message)s'
        self.log_name = os.path.join('./log/Business.log')
        self.logger = logging.getLogger('Sw_Script')
        self.logger.setLevel(logging.DEBUG)
        self.formatter = logging.Formatter(selfdef_fmt)

    def __console(self, level, message):
        fh = logging.FileHandler(self.log_name, 'a', encoding='utf-8')
        fh.setLevel(logging.DEBUG)
        fh.setFormatter(self.formatter)
        self.logger.addHandler(fh)

        ch = logging.StreamHandler()  # 创建一个StreamHandler,用于输出到控制台
        ch.setLevel(logging.DEBUG)
        ch.setFormatter(self.formatter)
        self.logger.addHandler(ch)

        if level == 'info':
            self.logger.info(message)
        elif level == 'debug':
            self.logger.debug(message)
        elif level == 'warning':
            self.logger.warning(message)
        elif level == 'error':
            self.logger.error(message)
        elif level == 'critical':
            self.logger.critical(message)

        self.logger.removeHandler(ch)
        self.logger.removeHandler(fh)
        fh.close()

    def debug(self, message):
        self.__console('debug', message)

    def info(self, message):
        self.__console('info', message)

    def warning(self, message):
        self.__console('warning', message)

    def error(self, message):
        self.__console('error', message)

    def critical(self, message):
        self.__console('critical',message)


log = Logger()



	 

	Read_source.py

	 
import csv


class Read_date(object):
    def __init__(self, path: str):
        self._path = path

    def get_entry_ip(self) -> list[str]:
        with open(self._path) as f:
            entry = csv.reader(f)
            ip = [ip[0] for ip in entry]
        return ip

    def get_entry_name(self) -> list[str]:
        with open(self._path) as f:
            entry = csv.reader(f)
            name = [name[1] for name in entry]
        return name

    def get_entry_passwd(self) -> list[str]:
        with open(self._path) as f:
            entry = csv.reader(f)
            passwd = [passwd[2] for passwd in entry]
        return passwd


login = Read_date('./source/entry_table.csv')


	 

	run.py

	 
import backup_Script
from backup_Script import log

if __name__ == "__main__":
    log.error('运行脚本')
    backup_Script.run()
    log.error('执行完成')


	 

	requirements.txt

	 
bcrypt==4.0.0
certifi==2022.6.15
cffi==1.15.1
charset-normalizer==2.1.1
cryptography==37.0.4
docopt==0.6.2
idna==3.3
logger==1.4
paramiko==2.11.0
pipreqs==0.4.11
pycparser==2.21
PyNaCl==1.5.0
requests==2.28.1
six==1.16.0
urllib3==1.26.12
yarg==0.1.9


	 

	　　审核编辑：汤梓红