0.前言
由于都是发自己的文章,所以也不会有那么多的内容发布,一周可能也就更新个两篇,如果工作忙,可能一周就一篇,但是文章都会很细致,暂时只更新学习笔记,至于挖洞思路,小菜鸟还不配。
0.1.免责声明
传播、利用本公众号剁椒鱼头没剁椒所提供的信息而造成的任何直接或者间接的后果及损失,均由使用者本人负责,公众号剁椒鱼头没剁椒及作者不为此承担任何责任,一旦造成后果请自行承担!如有侵权烦请告知,我们会立即删除并致歉。谢谢!
1. 前言
昨天我看到后台有人回复使用 python 编写交换机路由器配置文件,但是我的文章中都是安全,所以我再去年就写好了这个工具,这是用于我日常备份交换机路由器配置文件的,没办法项目里面有对设备的备份这项工作,但是交换机路由器设备太多了,如果我手动每一台登陆上去下载配置,估计要保存到吐,索性就开发了一个简单的工具,不过由于不符合公众号的属性所以就没发,但看到有人问,那我还是发一下吧。
源码下载:后台回复“自动备份312”,我把 exe 也添加进去了,怕有毒的隔离运行,我也不会闲着无聊加毒。
2. 前期准备
这里前期需要准备这些文件,不过部分文件是自动产生的,也并非需要手动创建,下面会提示,目前这个脚本仅支持华为和华三,思科,锐捷
重点:
注意需要将华为或华三设备的 FTP 功能开启,同时华为交换机配置名字为:vrpcfg.zip
,华三交换机配置名字为:startup.cfg
,所以当出现下载失败的时候,请第一时间确定配置名称是否为这两个,可能由于不同的型号或者版本导致配置名称不同出现下载失败。
downloaded_configs
:备份完文件保存的目录,无需创建,会自动生成。
File_backup.exe
:备份脚本,脚本名称无所谓,修改成什么都可以,这里我是制作成 exe 文件了。----这个可以不需要,这是我最近制作的。
File_backup.py
:是备份脚本的源码。
download_errors.log
:错误日志,无需创建,会自动生成。
IP.xlsx
:存放 IP 信息的,需要按照下图格式填写,名字可变
hosthu | username | password | device_type) |
---|---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
2.1. 脚本源代码
import osimport timeimport signalimport loggingimport argparsefrom ftplib import FTP, error_temp, error_permimport pandas as pdfrom concurrent.futures import ThreadPoolExecutor, as_completedfrom colorama import init, Fore, Styleimport threading# 初始化 coloramainit(autoreset=True)# 设置日志记录logging.basicConfig(filename='download_errors.log', level=logging.ERROR, format='%(asctime)s %(levelname)s: %(message)s')# 创建线程锁lock = threading.Lock()# 定义一个函数来验证文件是否成功下载defis_file_downloaded(local_filename, expected_size=None):if os.path.exists(local_filename): file_size = os.path.getsize(local_filename)if expected_size:return file_size >= expected_sizeelse:return file_size > 0returnFalse# 定义下载文件函数defdownload_file(ftp, filename, local_filename):try:with open(local_filename, 'wb') as local_file: ftp.retrbinary(f'RETR {filename}', local_file.write)# 验证文件是否成功下载if is_file_downloaded(local_filename): relative_path = os.path.relpath(local_filename, os.getcwd())with lock: print(Fore.GREEN + f'{filename} 已下载为 {relative_path}') # 下载成功时显示绿色else:raise Exception(f'{filename} 下载失败,文件大小为0')except error_perm as e: logging.error(f"FTP 权限错误,无法下载文件: {e}")raiseexcept Exception as e: logging.error(f"下载文件时出错: {e}")raise# 定义下载华为配置函数defdownload_huawei_config(host, username, password, download_dir, timeout=5): local_filename = os.path.join(download_dir, f'{host}_vrpcfg.zip') # 在 try 块外定义try: ftp = FTP(host, timeout=timeout) # 设置超时时间 ftp.login(user=username, passwd=password)with lock: print(Fore.YELLOW + f'连接华为设备: {host}') # 连接设备时显示黄色# 下载文件到指定目录 download_file(ftp, 'vrpcfg.zip', local_filename) ftp.quit()except error_temp as e:with lock: logging.error(f'下载华为配置时出错: {host} 连接超时或暂时不可用: {e}') print(Fore.RED + f"从 {host} 下载错误。查看日志了解更多详细信息。") # 错误时显示红色except ConnectionResetError as e:if is_file_downloaded(local_filename):# 忽略连接被强制关闭的错误,因为文件已成功下载with lock: print(Fore.CYAN + f"{host} 下载完成后连接被强制关闭,但文件已成功下载。")else: logging.error(f'下载华为配置时连接被强制关闭,且文件下载不完整: {host} 错误: {e}') print(Fore.RED + f"从 {host} 下载错误。查看日志了解更多详细信息。")except Exception as e:with lock: logging.error(f'下载华为配置时出错: {host}: {e}') print(Fore.RED + f"从 {host} 下载错误。查看日志了解更多详细信息。") # 错误时显示红色# 定义下载华三配置函数(与华为类似)defdownload_h3c_config(host, username, password, download_dir, timeout=5): local_filename = os.path.join(download_dir, f'{host}_startup.cfg') # 在 try 块外定义try: ftp = FTP(host, timeout=timeout) # 设置超时时间 ftp.login(user=username, passwd=password)with lock: print(Fore.YELLOW + f'连接H3C设备: {host}') # 连接设备时显示黄色# 下载文件到指定目录 download_file(ftp, 'startup.cfg', local_filename) ftp.quit()except error_temp as e:with lock: logging.error(f'下载H3C配置时出错: {host} 连接超时或暂时不可用: {e}') print(Fore.RED + f"从 {host} 下载错误。查看日志了解更多详细信息。") # 错误时显示红色except ConnectionResetError as e:if is_file_downloaded(local_filename):with lock: print(Fore.CYAN + f"{host} 下载完成后连接被强制关闭,但文件已成功下载。")else: logging.error(f'下载H3C配置时连接被强制关闭,且文件下载不完整: {host} 错误: {e}') print(Fore.RED + f"从 {host} 下载错误。查看日志了解更多详细信息。")except Exception as e:with lock: logging.error(f'下载H3C配置时出错: {host}: {e}') print(Fore.RED + f"从 {host} 下载错误。查看日志了解更多详细信息。") # 错误时显示红色# 从 Excel 文件中读取设备信息defload_config_from_excel(file_path): df = pd.read_excel(file_path) devices = df.to_dict(orient='records')return devices# 根据设备类型调用相应的下载函数defdownload_config(device, download_dir): timeout = 30# 设置超时时间# 防止设备类型是浮点数或其他类型,先转换为字符串 device_type = str(device['device_type']).lower() if'device_type'in device elseNoneif device_type == 'huawei': download_huawei_config(device['host'], device['username'], device['password'], download_dir, timeout)elif device_type == 'h3c': download_h3c_config(device['host'], device['username'], device['password'], download_dir, timeout)else:with lock: logging.error(f"未知设备类型: {device['device_type']},主机: {device['host']}") print(Fore.RED + f"未知设备类型: {device['device_type']},主机: {device['host']}") # 错误时显示红色# 捕获 Ctrl+C 信号defsignal_handler(sig, frame):with lock: logging.warning("程序被用户中断") print(Fore.RED + "n过程中断,退出……") exit(0)# 主函数defmain(excel_file, num_threads):# 清空日志文件 open('download_errors.log', 'w').close() # 清空日志文件# 记录开始时间 start_time = time.time()# 创建下载目录 download_dir = os.path.join(os.getcwd(), 'downloaded_configs') os.makedirs(download_dir, exist_ok=True)# 从 Excel 加载设备配置 devices = load_config_from_excel(excel_file)# 使用多线程下载配置with ThreadPoolExecutor(max_workers=num_threads) as executor: futures = [executor.submit(download_config, device, download_dir) for device in devices]# 捕获所有任务完成的情况for future in as_completed(futures):try: future.result()except Exception as e:with lock: logging.error(f"线程错误: {e}")# 计算并输出运行时间 end_time = time.time() elapsed_time = end_time - start_timewith lock: print(Fore.YELLOW + f"n下载完成耗时: {elapsed_time:.2f} 秒。") # 下载完成时显示黄色if __name__ == '__main__': banner = """程序名称: 从华为或H3C交换机路由器下载配置使用说明: - 首先交换机路由器需要开启FTP功能 - 设备IP,首行填写:host、username、password、device_type 二行填写:127.0.0.1、admin、123456、huawei/h3c - 线程数量支持单个整数,默认1线程,线程过高会存在误报,同时错误日志中可自行确认是误报还是连通性问题 - 结果保存在当前目录下的downloaded_configs目录中,文件格式:{ip}_startup.cfg/{ip}_vrpcfg.zip - 错误日志保存在download_errors.log """ print(banner)# 解析命令行参数 parser = argparse.ArgumentParser() parser.add_argument('-f', dest='excel_file', type=str, required=True, help='保存设备信息的Excel文件的路径。') # 设置为必选参数 parser.add_argument('-t', dest='threads', type=int, default=1, help='要使用的并发线程数。默认值是1。') args = parser.parse_args()# 捕获 Ctrl+C 信号 signal.signal(signal.SIGINT, signal_handler)# 调用主函数 main(args.excel_file, args.threads)
2.2. 运行效果
网络通信正常的情况下,我测试了 62 台设备 5 线程约 23 秒完成,其中有一台设备不通信异常约 4 秒。
File_backup.py -f IP.xlsx(IP地址表) -t 2(线程数,默认为1,建议5线程)
2.3. 错误日志
这里会提示是连接问题,还是下载错误,如果 FTP 没开的话也会出现连接问题。
原文始发于微信公众号(剁椒鱼头没剁椒):Python开发交换路由设备配置备份工具
- 左青龙
- 微信扫一扫
-
- 右白虎
- 微信扫一扫
-
评论