如何通過Python腳本批量采集華為交換機(jī)配置

實驗?zāi)康?/p>

通過學(xué)習(xí)python網(wǎng)絡(luò)運維自動化減少重復(fù)無意義的工作量，本次實驗雖然只是一條命令，通過display current-configuration采集設(shè)備配置，但是在有大量設(shè)備需要采集配置文件時，又無相應(yīng)的網(wǎng)管平臺時，小而便捷的python成為了一種選擇，而且可以python腳本可以根據(jù)需要自行增加需要的命令。

設(shè)備環(huán)境

通過ENSP模擬華為交換機(jī)，橋接云跟主機(jī)通信，SSH配置可達(dá)。

設(shè)備配置命令：

sysnameSW1
#不同設(shè)備IP不同，sysname不同，其他配置一致
interfaceVlanif1
ipaddress192.168.111.10255.255.255.0
#
aaa
local-useradminpasswordcipher

	

	其余兩臺按這個模板配置


	橋接云配置

	完成設(shè)備配置后測試SSH連接是否正常


	通過CRT測試，三臺設(shè)備都能正常登錄

	python環(huán)境

	首先看一下項目的結(jié)構(gòu)


	目錄說明

	backup_Script包是本次實驗的主要代碼，分為了三個模塊來寫的。

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

	log是日志文件保存目錄。

	source是中存在一個entry_table.csv，這是填寫登錄IP跟用戶名的。

	

	不能更改列數(shù)，更改IP跟用戶名密碼的時候，就按照這個來更改

	requirements.txt里面內(nèi)容是項目的依賴包及其對應(yīng)版本號的信息列表，即項目依賴關(guān)系清單，其作用是用來重新構(gòu)建項目所需要的運行環(huán)境依賴。

	run.py作為本次實驗的入口文件，在運行項目的時候就運行該文件

	環(huán)境搭建

	：版本Python 3.10.2

	環(huán)境的搭建就從創(chuàng)建虛擬環(huán)境開始寫，python安裝之類的參考官方網(wǎng)站


	在最初項目的結(jié)構(gòu)是這樣的，然后進(jìn)入到text目錄下構(gòu)建虛擬環(huán)境。

	首先進(jìn)入項目的目錄，創(chuàng)建虛擬環(huán)境

	
#ven是可變的
python-mvenvven


	

	

	構(gòu)建完成后在次查看目錄，會發(fā)現(xiàn)目錄中多了一個目錄ven，這個ven就是所創(chuàng)建的虛擬環(huán)境。


	當(dāng)然，虛擬環(huán)境創(chuàng)建好后不代表就結(jié)束了，還需要進(jìn)入到虛擬環(huán)境中，稱為激活虛擬環(huán)境

	
#通過該方法激活虛擬環(huán)境，激活后會有一個括號在前面
venScriptsactivate
#以下是激活后的顯示
(ven)D:	ext>
#激活完成更新以下pip庫
python.exe-mpipinstall--upgradepip
#安裝所需依賴
pipinstall-rrequirements.txt
#通過入口文件啟用
run.py


	

	


	從控制臺輸出的日志可以看出，腳本已經(jīng)成功運行結(jié)束，那么查看一下相關(guān)的目錄是否存在這文件


	查看Dest目錄，已經(jīng)存在三個文件，打開看一下是否已經(jīng)獲取到內(nèi)容


	查看日志文件夾

	

	也存在相關(guān)的日志

	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('登錄失敗，詳細(xì)查看文件log/switch_not_reachable.txt為不可達(dá)，switch_with_authentication_issue.txt為認(rèn)證失敗')
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"用戶認(rèn)證失敗的{ip}.")
switch_with_authentication_issue.append(ip)
exceptsocket.error:
log.critical(f"{ip}不可達(dá)，請檢查網(wǎng)絡(luò).")
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()#創(chuàng)建一個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('執(zhí)行完成')


	

	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


	

	審核編輯：湯梓紅