在游水的时候看到 @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"}
]
评论 (0)