如何通过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 # 激活完成更新以下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 loginfrom .logger import logimport paramikoimport timeimport socketimport reimport syssys.path.append('backup_script')ssh_client = noneswitch_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 loggingimport osclass 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 csvclass 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 passwdlogin = read_date('./source/entry_table.csv')
run.py
import backup_scriptfrom backup_script import logif __name__ == __main__: log.error('运行脚本') backup_script.run() log.error('执行完成')
requirements.txt
bcrypt==4.0.0certifi==2022.6.15cffi==1.15.1charset-normalizer==2.1.1cryptography==37.0.4docopt==0.6.2idna==3.3logger==1.4paramiko==2.11.0pipreqs==0.4.11pycparser==2.21pynacl==1.5.0requests==2.28.1six==1.16.0urllib3==1.26.12yarg==0.1.9
基于AVR单片机的新型防汽车追尾安全装置设计
日本研制出基于石墨烯的高度集成、高速硅基芯片黑体发射器
宝砾微PL88101 SOT23-6L降压型开关稳压器概述
最新日本NTT发明鞋发电机未来可为手机供电
采用被动探测体制的识别技术实现分布式智能无线信息侦察传感系统设计
如何通过Python脚本批量采集华为交换机配置
西门子S7-200系列PLC对汽车焊接机器人控制研究
如何画好原理图，画原理图的7个非常见技巧
附带ON/OFF开关，300mA的高速LDO电压调整器
闪存真的要降价了 东芝成功研发世界首款4比特3D QLC闪存 大容量硬盘将更便宜
BGA的过孔开窗设计方案详解
苹果春季发布会将推出2021款iPad Pro
物联网onenet云平台的成功上线为5G新技术赋能智慧校园建设打下基础
苹果三星纷纷入局无线充电领域,无线充电将引来爆发式增长
关于伺服拉力试验机的工作原理，您知道如何租赁测试设备吗？
同心筑生态，前路共飞腾——飞腾CPU首届生态合作伙伴大会在京隆重举行！
超二代微光像增强器性能随工作时间的变化规律研究
魅族Pro7、魅族Max2最新消息：魅蓝E2才推出，魅族全年的产品规划全曝光！魅族Pro7、魅族Max2都在这
激光雷达“3.0时代”来临了吗？
盘点RISC-V MCU厂商，爱普特为何能在市场占有率拔得头筹？
通过学习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 # 激活完成更新以下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 loginfrom .logger import logimport paramikoimport timeimport socketimport reimport syssys.path.append('backup_script')ssh_client = noneswitch_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 loggingimport osclass 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 csvclass 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 passwdlogin = read_date('./source/entry_table.csv')
run.py
import backup_scriptfrom backup_script import logif __name__ == __main__: log.error('运行脚本') backup_script.run() log.error('执行完成')
requirements.txt
bcrypt==4.0.0certifi==2022.6.15cffi==1.15.1charset-normalizer==2.1.1cryptography==37.0.4docopt==0.6.2idna==3.3logger==1.4paramiko==2.11.0pipreqs==0.4.11pycparser==2.21pynacl==1.5.0requests==2.28.1six==1.16.0urllib3==1.26.12yarg==0.1.9
基于AVR单片机的新型防汽车追尾安全装置设计
日本研制出基于石墨烯的高度集成、高速硅基芯片黑体发射器
宝砾微PL88101 SOT23-6L降压型开关稳压器概述
最新日本NTT发明鞋发电机未来可为手机供电
采用被动探测体制的识别技术实现分布式智能无线信息侦察传感系统设计
如何通过Python脚本批量采集华为交换机配置
西门子S7-200系列PLC对汽车焊接机器人控制研究
如何画好原理图，画原理图的7个非常见技巧
附带ON/OFF开关，300mA的高速LDO电压调整器
闪存真的要降价了 东芝成功研发世界首款4比特3D QLC闪存 大容量硬盘将更便宜
BGA的过孔开窗设计方案详解
苹果春季发布会将推出2021款iPad Pro
物联网onenet云平台的成功上线为5G新技术赋能智慧校园建设打下基础
苹果三星纷纷入局无线充电领域,无线充电将引来爆发式增长
关于伺服拉力试验机的工作原理，您知道如何租赁测试设备吗？
同心筑生态，前路共飞腾——飞腾CPU首届生态合作伙伴大会在京隆重举行！
超二代微光像增强器性能随工作时间的变化规律研究
魅族Pro7、魅族Max2最新消息：魅蓝E2才推出，魅族全年的产品规划全曝光！魅族Pro7、魅族Max2都在这
激光雷达“3.0时代”来临了吗？
盘点RISC-V MCU厂商，爱普特为何能在市场占有率拔得头筹？