优化 Python 批量备份交换机配置-重构部分功能

TanHaX
2024-02-20 / 0 评论 / 46 阅读 / 正在检测是否收录...

在游水的时候看到 @winty 大佬 被 @aa7088414 大佬重构的代码后 感觉可以再优化那么一丢丢。

主要修改以下几点:

  1. 代码结构整理:对原始代码进行了逻辑重构和优化,将部分重复代码提取成函数进行复用,简化代码结构。
  2. 错误处理:在连接设备时添加了异常处理,并记录连接失败的设备信息。
  3. 文件操作:使用了os.makedirs替换了os.path.joinos.makedirs组合,同时对异常进行了处理。
  4. 数据读取:使用openpyxl库的iter_rows方法直接获取单元格值,避免了通过循环逐一获取单元格值的冗余操作。
  5. 日志记录:使用logging库记录了操作日志,并对错误进行详细记录。
  6. 代码细节:对变量和字典键值等进行了类型转换和判空处理。
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

打赏

评论 (0)

取消