尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

别再手动备份了!用Python脚本批量导出华为/华三交换机配置(附完整代码)

别再手动备份了!用Python脚本批量导出华为/华三交换机配置(附完整代码) 华为/华三交换机自动化配置备份实战指南Python脚本开发与运维整合每次手动登录几十台交换机备份配置的日子该结束了。作为一名长期奋战在一线的网络工程师我深知这种重复性工作不仅耗时费力还容易因人为疏忽导致备份遗漏或错误。本文将分享一套经过实战检验的Python自动化解决方案帮助您实现华为、华三交换机配置的智能备份与管理。1. 自动化备份的核心价值与设计思路传统手动备份方式存在三大痛点操作繁琐耗时、容易遗漏设备、备份文件管理混乱。我们设计的自动化系统需要解决这些问题同时考虑企业级环境中的实际需求。关键设计原则全自动执行从设备登录到文件保存全程无需人工干预智能异常处理网络中断、认证失败等情况自动重试并记录规范化存储按时间、设备类型自动分类存档可扩展架构方便添加对新设备型号的支持实际测试数据显示对于100台设备的备份任务手动操作平均耗时约5小时3分钟/台自动化脚本耗时约25分钟包括异常重试时间2. 环境准备与依赖配置2.1 Python环境搭建推荐使用Python 3.8版本这是目前最稳定的选择。使用虚拟环境可以避免依赖冲突python -m venv switch_backup source switch_backup/bin/activate # Linux/Mac # 或者 switch_backup\Scripts\activate # Windows2.2 必需库安装与功能说明库名称版本要求功能用途替代方案paramiko≥2.9.0SSH连接与设备交互netmikopandas≥1.3.0设备清单Excel文件处理openpyxl直接操作openpyxl≥3.0.0Excel文件读写支持xlrd/xlwtpython-dotenv≥0.19.0敏感信息管理直接使用os.environ安装命令pip install paramiko pandas openpyxl python-dotenv提示生产环境中建议将密码等敏感信息存储在.env文件中而非直接写在脚本或Excel里3. 设备信息管理最佳实践规范的设备信息管理是自动化备份的基础。我们推荐使用结构化的Excel表格进行管理设备信息表示例结构IP地址设备命名设备型号管理员账号密码厂商类型备用管理IP192.168.1.1Core-SW1S5735-LadminxxxxxxHuawei10.1.1.1192.168.1.2Access-1S5130-EInetworkxxxxxxH3C高级管理技巧使用Excel数据验证确保厂商类型等字段的规范性为关键字段添加批注说明如密码更新周期设置条件格式标记异常IP地址使用冻结窗格方便查看大量设备设备信息读取代码优化版def read_device_info(file_pathNone): 智能读取设备信息表支持自动路径查找 if not file_path: desktop os.path.join(os.path.expanduser(~), Desktop) for fname in os.listdir(desktop): if 交换机设备信息 in fname and fname.endswith((.xlsx, .xls)): file_path os.path.join(desktop, fname) break if not file_path or not os.path.exists(file_path): raise FileNotFoundError(设备信息表未找到) df pd.read_excel(file_path) required_cols [IP地址, 设备命名, 设备型号, 管理员账号, 密码] if not all(col in df.columns for col in required_cols): missing set(required_cols) - set(df.columns) raise ValueError(f缺少必要列: {missing}) # 数据清洗 df[厂商类型] df[厂商类型].str.upper().str.strip() return df4. 核心备份功能实现与优化4.1 增强型SSH连接管理基础SSH连接存在超时、中断等问题我们实现带重试机制的连接方式class SwitchConnection: def __init__(self, ip, username, password, timeout30, retries3): self.ip ip self.cred (username, password) self.timeout timeout self.retries retries self.client None def __enter__(self): for attempt in range(1, self.retries 1): try: self.client paramiko.SSHClient() self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.client.connect( self.ip, usernameself.cred[0], passwordself.cred[1], timeoutself.timeout, banner_timeout60 ) return self.client except Exception as e: if attempt self.retries: raise time.sleep(2 ** attempt) # 指数退避 def __exit__(self, exc_type, exc_val, exc_tb): if self.client: self.client.close()4.2 配置备份流程优化标准备份流程存在等待时间过长或不足的问题我们实现动态等待机制def export_config(conn, device_info, folder_path): 智能配置备份自动适配不同厂商设备 vendor device_info[厂商类型].upper() with conn.invoke_shell() as shell: # 厂商特定初始化 if vendor HUAWEI: shell.send(screen-length 0 temporary\n) cmd display current-configuration\n terminator return elif vendor H3C: shell.send(screen-length disable\n) cmd display current-configuration\n terminator time.sleep(1) shell.recv(65535) # 清空缓冲区 # 发送配置收集命令 shell.send(cmd) # 动态等待输出完成 output start_time time.time() while time.time() - start_time 120: # 最大等待2分钟 time.sleep(2) if shell.recv_ready(): part shell.recv(65535).decode(utf-8) output part if terminator in output[-100:]: # 检查结束标记 break # 规范化保存 timestamp datetime.now().strftime(%Y%m%d%H%M%S) base_name f{device_info[设备命名]}_{vendor}_{timestamp} # 保存为文本和配置两种格式 for ext, content in [(.txt, output), (.cfg, output)]: file_path os.path.join(folder_path, f{base_name}{ext}) with open(file_path, w, encodingutf-8) as f: f.write(content) return True4.3 异常处理与日志记录完善的异常处理是生产环境应用的关键def backup_device(device_info, folder_path): 带异常处理和日志记录的设备备份 log_entry { timestamp: datetime.now().isoformat(), device: device_info[设备命名], ip: device_info[IP地址], status: started, error: None } try: with SwitchConnection( device_info[IP地址], device_info[管理员账号], device_info[密码] ) as conn: result export_config(conn, device_info, folder_path) log_entry[status] success if result else partial except paramiko.AuthenticationException: log_entry.update({ status: failed, error: Authentication failed }) except socket.timeout: log_entry.update({ status: failed, error: Connection timeout }) except Exception as e: log_entry.update({ status: failed, error: str(e) }) # 写入日志文件 log_file os.path.join(folder_path, backup_log.json) with open(log_file, a, encodingutf-8) as f: f.write(json.dumps(log_entry) \n) return log_entry[status] success5. 生产环境集成方案5.1 Windows计划任务配置对于Windows服务器环境可以通过计划任务实现定期自动备份创建运行脚本的批处理文件run_backup.batecho off set PYTHONPATHC:\Python38 call C:\Python38\python.exe C:\scripts\switch_backup.py使用schtasks创建计划任务$action New-ScheduledTaskAction -Execute C:\scripts\run_backup.bat $trigger New-ScheduledTaskTrigger -Weekly -DaysOfWeek Monday -At 2am Register-ScheduledTask -TaskName SwitchConfigBackup -Action $action -Trigger $trigger5.2 Jenkins持续集成方案在DevOps环境中可以通过Jenkins实现更灵活的备份管理Jenkinsfile配置示例pipeline { agent any triggers { cron(0 2 * * 1) // 每周一凌晨2点 } stages { stage(Backup) { steps { script { try { bat python D:\\scripts\\switch_backup.py emailext body: 交换机配置备份成功, subject: 备份成功通知, to: teamexample.com } catch (e) { emailext body: 备份失败: ${e}, subject: 备份失败警报, to: adminexample.com error 备份失败 } } } } stage(Archive) { steps { archiveArtifacts artifacts: backups/**, fingerprint: true } } } }5.3 备份文件生命周期管理长期积累的备份文件会占用大量存储空间需要制定保留策略def cleanup_old_backups(root_dir, keep_days30): 自动清理超过指定天数的备份文件 cutoff time.time() - keep_days * 86400 for dirpath, _, filenames in os.walk(root_dir): for fname in filenames: file_path os.path.join(dirpath, fname) if os.stat(file_path).st_mtime cutoff: try: os.remove(file_path) except Exception as e: print(f删除失败 {file_path}: {e}) # 删除空目录 for dirpath, dirnames, _ in os.walk(root_dir, topdownFalse): for dirname in dirnames: full_path os.path.join(dirpath, dirname) if not os.listdir(full_path): os.rmdir(full_path)6. 高级功能扩展6.1 配置差异比较定期备份的价值在于能够追踪配置变更实现配置漂移检测def compare_configs(old_file, new_file): 使用difflib比较两个配置文件的差异 with open(old_file, r, encodingutf-8) as f: old_lines f.readlines() with open(new_file, r, encodingutf-8) as f: new_lines f.readlines() differ difflib.HtmlDiff() return differ.make_file(old_lines, new_lines, old_file, new_file)6.2 多线程并行备份对于大规模网络环境串行备份效率太低可以使用线程池加速from concurrent.futures import ThreadPoolExecutor, as_completed def parallel_backup(devices, max_workers10): 使用线程池并行备份多台设备 folder_path create_folder() results {} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_device { executor.submit(backup_device, device, folder_path): device[设备命名] for _, device in devices.iterrows() } for future in as_completed(future_to_device): device_name future_to_device[future] try: results[device_name] future.result() except Exception as e: results[device_name] str(e) # 生成汇总报告 generate_report(results, folder_path) return folder_path6.3 备份完整性验证确保备份的配置完整可用def validate_config(content, vendor): 验证配置内容是否完整 required_sections { HUAWEI: [sysname, interface, vlan], H3C: [sysname, interface, vlan] } missing [] for section in required_sections[vendor]: if f[{section}] not in content and f{section} not in content: missing.append(section) if missing: raise ValueError(f配置缺少关键部分: {missing}) return True7. 安全增强措施自动化备份系统需要特别注意安全性安全最佳实践使用SSH密钥认证替代密码如设备支持配置最小权限账户进行备份加密存储设备凭证信息设置备份文件的访问权限控制定期轮换备份存储的加密密钥凭证管理示例from cryptography.fernet import Fernet class CredentialManager: def __init__(self, key_file.encryption_key): if not os.path.exists(key_file): self.key Fernet.generate_key() with open(key_file, wb) as f: f.write(self.key) else: with open(key_file, rb) as f: self.key f.read() self.cipher Fernet(self.key) def encrypt(self, text): return self.cipher.encrypt(text.encode()).decode() def decrypt(self, encrypted_text): return self.cipher.decrypt(encrypted_text.encode()).decode() # 使用示例 manager CredentialManager() encrypted_pw manager.encrypt(my_password) decrypted_pw manager.decrypt(encrypted_pw)
返回列表