优化 Python 批量备份交换机配置-重构部分功能 在游水的时候看到 @winty 大佬 被 @aa7088414 大佬重构的代码后 感觉可以再优化那么一丢丢。主要修改以下几点：代码结构整理：对原始代码进行了逻辑重构和优化，将部分重复代码提取成函数进行复用，简化代码结构。错误处理：在连接设备时添加了异常处理，并记录连接失败的设备信息。文件操作：使用了os.makedirs替换了os.path.join和os.makedirs组合，同时对异常进行了处理。数据读取：使用openpyxl库的iter_rows方法直接获取单元格值，避免了通过循环逐一获取单元格值的冗余操作。日志记录：使用logging库记录了操作日志，并对错误进行详细记录。代码细节：对变量和字典键值等进行了类型转换和判空处理。import json import logging import time from datetime import datetime import os import re from netmiko import ConnectHandler, exceptions from openpyxl import load_workbook # 取消以下注释开启调试模式 # logging.basicConfig(level=logging.DEBUG) # 读取excel内设备列表信息 def check_and_get_dev_list(filename, sheet_name): excel_information = [] wb = load_workbook(filename) sh = wb[sheet_name] # 获取最大行数 rows = list(sh.iter_rows(values_only=True)) sheet_header = [cell for cell in rows[0]] data = [dict(zip(sheet_header, row)) for row in rows[1:]] # 去除ip为空的数据 data = [i for i in data if i.get('ip') is not None] return data # 获取excel数据并整合成dev字典 def get_dev(): res = check_and_get_dev_list('./resource.xlsx', 'Sheet1') devices = [] for i in res: if i['protocol'] == 'telnet': i['type'] = i['type'] + '_telnet' dev = { 'device_type': str(i['type']), 'host': str(i['ip']), 'username': str(i['username']), 'password': str(i['password']), 'secret': str(i['enpassword']) if i['enpassword'] else None, 'port': str(i['port']), 'session_log': 'session.log' } devices.append(dev) return devices def get_cmds(): with open("./dis_cmds.conf", 'r', encoding='utf-8') as f: js = json.load(f) return js def devices_confbak(devices=None): if devices is None: devices = [] current_date = datetime.now().strftime("%Y-%m-%d") path = f'./conf_bak/{current_date}' try: os.makedirs(path, exist_ok=True) except OSError as e: logging.error(f"Failed to create directory: {path}, error: {e}") failed_ips = [] for dev in devices: try: with ConnectHandler(**dev) as conn: print('\n----------成功登录到：' + dev['host'] + '----------') if dev['secret'] and dev['secret'] != "None": try: conn.enable() except Exception as e: logging.error("enable登录失败: %s", e) pwd = f"{path}/{dev['host']}" os.makedirs(pwd, exist_ok=True) dis_cmds = get_cmds() for cmd_string_dict in dis_cmds: print(f"正在获取：{cmd_string_dict['name']}") time.sleep(0.1) output = None for dev_type, command in cmd_string_dict.items(): if dev['device_type'].startswith(dev_type): output = conn.send_command(command_string=command) if output is None: print(f"error：未找到设备类型为{dev['device_type']}的{cmd_string_dict['name']}预设命令，请检查并修改dis_cmds.conf文件") continue fname = f"{dev['host']}_{cmd_string_dict['name']}.txt" file_path = f"{pwd}/{fname}" with open(file_path, mode='w', encoding='utf8') as f: try: f.write(output) print(f"备份{fname}成功！") except PermissionError: print("*****-无写入权限，请将文件夹赋予读写权限-*****") except (exceptions.NetmikoAuthenticationException, exceptions.NetmikoTimeoutException, exceptions.ReadTimeout) as e: logging.error(f"连接设备{dev['host']}失败: {e}") failed_ips.append(dev['host']) except Exception as e: logging.error(f"发生了意料外的错误：{e}") failed_ips.append(dev['host']) if failed_ips: print("\n以下设备连接失败,请检查：") for ip in failed_ips: print(ip) return 0 def devices_autocheck(devices='', cmd=''): results = [] try: for device in devices: with ConnectHandler(**device) as conn: conn.enable() print('正在巡检：' + device['host'] + ' ...') result = [device['host'], device['device_type']] for item in cmd: if 'cisco_ios' in device['device_type']: output = conn.send_command(command_string=str(item['cisco'])) elif any(x in device['device_type'] for x in ['huawei', 'hp_comware']): conn.send_command(command_string='sys', expect_string=']') output = conn.send_command(command_string=str(item['huawei'])) result.append(output) results.append(result) except (exceptions.NetmikoAuthenticationException, exceptions.NetmikoTimeoutException, exceptions.ReadTimeout) as e: logging.error(f"连接设备{device['host']}失败: {e}") return results def get_mem(memstr, devtype=''): if 'cisco' in devtype: total_match = re.search(r'Processor Pool Total:\s+(\d+)', memstr) used_match = re.search(r'Used:\s+(\d+)', memstr) if total_match and used_match: total = int(total_match.group(1)) used = int(used_match.group(1)) percentage = used / total * 100 return f"{percentage:.0f}%" elif 'huawei' in devtype: match = re.search(r"Memory Using Percentage Is:\s*(\d+)%", memstr) if match: return match.group(1) + '%' return "No match found." def get_cpu(cpustr, devtype=''): if 'cisco' in devtype: match = re.search(r"CPU utilization for five seconds: (\d+)%", cpustr) if match: return match.group(1) + '%' elif 'huawei' in devtype: match = re.search(r"\b(\d+(\.\d+)?)%.*?\bMax", cpustr) if match: return match.group(1) + '%' return "No match found." if __name__ == '__main__': while True: print("\n##############################################\n") print("1：批量备份交换机信息与配置") print("2：批量巡检交换机设备") print("0：退出") option = input("请输入需要的操作编号：") if option == '1': dev = get_dev() devices_confbak(devices=dev) continue elif option == '2': cmds = [ {'cisco': 'show clock', 'huawei': 'display clock'}, {'cisco': 'show env power', 'huawei': 'display power'}, {'cisco': 'show env fan', 'huawei': 'display fan'}, {'cisco': 'show env temperature status', 'huawei': 'display environment'}, {'cisco': 'show processes cpu', 'huawei': 'display cpu-usage'}, {'cisco': 'show processes memory', 'huawei': 'display memory-usage'}, ] dev = get_dev() checkres = devices_autocheck(dev, cmds) for res in checkres: print('\n+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++') print(res[0] + '-巡检结果：') print('\n时钟：\n' + res[2]) print('电源：\n' + res[3]) print('风扇：\n' + res[4]) if 'Unrecognized command' in res[5]: print('温度：\n该设备不支持获取此数据！') else: print('温度：\n' + res[5]) print('CPU利用率：\n' + get_cpu(res[6], res[1])) print('内存利用率：\n' + get_mem(res[7], res[1])) print('\n+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++') continue elif option == '0': break else: print("请输入正确的编号！") 注：python 的版本为3.9以下是配套的文件dis_cmds.conf（修改时引号必须为英文双引号）：[ {"name": "查看版本", "hp_comware": "display version", "huawei": "display version", "cisco_ios": "show version"}, {"name": "查看时间", "hp_comware": "display clock", "huawei": "display clock", "cisco_ios": "show clock"}, {"name": "查看序列号", "hp_comware": "display device ma", "huawei": "display device manufacture-info", "cisco_ios": "show inventory"}, {"name": "查看风扇", "hp_comware": "display fan", "huawei": "display fan", "cisco_ios": "show environment"}, {"name": "查看电源状态", "hp_comware": "display power", "huawei": "display power", "cisco_ios": "show environment power"}, {"name": "查看CPU利用率", "hp_comware": "display cpu-usage", "huawei": "display cpu-usage", "cisco_ios": "show processes cpu"}, {"name": "查看内存利用率", "hp_comware": "display memory", "huawei": "display memory-usage", "cisco_ios": "show processes memory"}, {"name": "查看VLAN配置", "hp_comware": "display vlan", "huawei": "display vlan", "cisco_ios": "show vlan"}, {"name": "查看二层接口状态", "hp_comware": "display interface brief", "huawei": "display interface brief", "cisco_ios": "show interfaces switchport"}, {"name": "查看三层接口IP", "hp_comware": "display ip interface brief", "huawei": "display ip interface brief", "cisco_ios": "show ip interface brief"}, {"name": "查看ACL控制列表", "hp_comware": "display acl", "huawei": "display acl all", "cisco_ios": "show access-lists"}, {"name": "查看路由表", "hp_comware": "display ip routing-table", "huawei": "display ip routing-table", "cisco_ios": "show ip route"}, {"name": "查看ARP表", "hp_comware": "display arp", "huawei": "display arp", "cisco_ios": "show arp"}, {"name": "查看MAC表", "hp_comware": "display mac-address", "huawei": "display mac-address", "cisco_ios": "show mac address-table"}, {"name": "查看日志", "hp_comware": "display logbuffer", "huawei": "display logbuffer", "cisco_ios": "show logging"}, {"name": "查看所有配置", "hp_comware": "display current-configuration", "huawei": "display current-configuration", "cisco_ios": "show running-config"} ]{cloud title="交换机批量备份_源码" type="lz" url="https://tanghan.lanzoul.com/ixtqL1ou6mla" password="ao5m"/}