如何通过Python脚本批量采集华为交换机配置-电子发烧友网

实验目的

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

设备环境

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

设备配置命令：

sysnameSW1
#不同设备IP不同，sysname不同，其他配置一致
interfaceVlanif1
ipaddress192.168.111.10255.255.255.0
#
aaa
local-useradminpasswordcipher

	

	其余两台按这个模板配置


	桥接云配置

	完成设备配置后测试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-mvenvven


	

	

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


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

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


	

	


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


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


	查看日志文件夹

	

	也存在相关的日志

	python源码分享

	一、backup_Script包中的源码

	_init_.py

	
from.Read_sourceimportlogin
from.Loggerimportlog
importparamiko
importtime
importsocket
importre
importsys

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


deflogon_failed():
globalswitch_with_authentication_issue
globalswitch_not_reachable
ifswitch_with_authentication_issueandswitch_not_reachableisNone:
log.error('无登录失败记录')
else:
log.error('登录失败，详细查看文件log/switch_not_reachable.txt为不可达，switch_with_authentication_issue.txt为认证失败')
withopen('./log/switch_with_authentication_issue.txt','w')asf:
f.write('
'.join(switch_with_authentication_issue))
withopen('./log/switch_not_reachable.txt','w')asf:
f.write('
'.join(switch_not_reachable))


defrun():
globalssh_client
globalswitch_with_authentication_issue
globalswitch_not_reachable
ip_list=login.get_entry_ip()
name_list=login.get_entry_name()
passwd_list=login.get_entry_passwd()
regex='sysname.*'
foriinzip(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-length0temporary
")
command.send("displaycurrent-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()
exceptparamiko.ssh_exception.AuthenticationException:
log.critical(f"用户认证失败的{ip}.")
switch_with_authentication_issue.append(ip)
exceptsocket.error:
log.critical(f"{ip}不可达，请检查网络.")
switch_not_reachable.append(ip)
ssh_client.close()
logon_failed()



	

	Logger.py

	
importlogging
importos


classLogger:
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)

iflevel=='info':
self.logger.info(message)
eliflevel=='debug':
self.logger.debug(message)
eliflevel=='warning':
self.logger.warning(message)
eliflevel=='error':
self.logger.error(message)
eliflevel=='critical':
self.logger.critical(message)

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

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

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

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

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

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


log=Logger()



	

	Read_source.py

	
importcsv


classRead_date(object):
def__init__(self,path:str):
self._path=path

defget_entry_ip(self)->list[str]:
withopen(self._path)asf:
entry=csv.reader(f)
ip=[ip[0]foripinentry]
returnip

defget_entry_name(self)->list[str]:
withopen(self._path)asf:
entry=csv.reader(f)
name=[name[1]fornameinentry]
returnname

defget_entry_passwd(self)->list[str]:
withopen(self._path)asf:
entry=csv.reader(f)
passwd=[passwd[2]forpasswdinentry]
returnpasswd


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


	

	run.py

	
importbackup_Script
frombackup_Scriptimportlog

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


	

	审核编辑：汤梓红