實驗目的
通過學習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
審核編輯:湯梓紅
-
華為
+關注
關注
216文章
34534瀏覽量
253006 -
交換機
+關注
關注
21文章
2656瀏覽量
100180 -
python
+關注
關注
56文章
4807瀏覽量
85039 -
腳本
+關注
關注
1文章
392瀏覽量
14938 -
華為交換機
+關注
關注
0文章
13瀏覽量
6331
原文標題:如何通過Python腳本批量采集華為交換機配置,值得每位網絡工程師參考!
文章出處:【微信號:網絡技術干貨圈,微信公眾號:網絡技術干貨圈】歡迎添加關注!文章轉載請注明出處。
發布評論請先 登錄
相關推薦
評論