别再手动备份了！用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 # Windows

2.2 必需库安装与功能说明

安装命令：

pip install paramiko pandas openpyxl python-dotenv

提示：生产环境中建议将密码等敏感信息存储在.env文件中，而非直接写在脚本或Excel里

3. 设备信息管理最佳实践

规范的设备信息管理是自动化备份的基础。我们推荐使用结构化的Excel表格进行管理：

设备信息表示例结构：

高级管理技巧：

1. 使用Excel数据验证确保厂商类型等字段的规范性
2. 为关键字段添加批注说明（如密码更新周期）
3. 设置条件格式标记异常IP地址
4. 使用冻结窗格方便查看大量设备

设备信息读取代码优化版：

def read_device_info(file_path=None): """智能读取设备信息表，支持自动路径查找""" 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 df

4. 核心备份功能实现与优化

4.1 增强型SSH连接管理

基础SSH连接存在超时、中断等问题，我们实现带重试机制的连接方式：

class SwitchConnection: def __init__(self, ip, username, password, timeout=30, retries=3): 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, username=self.cred[0], password=self.cred[1], timeout=self.timeout, banner_timeout=60 ) 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', encoding='utf-8') as f: f.write(content) return True

4.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', encoding='utf-8') as f: f.write(json.dumps(log_entry) + '\n') return log_entry['status'] == 'success'

5. 生产环境集成方案

5.1 Windows计划任务配置

对于Windows服务器环境，可以通过计划任务实现定期自动备份：

1. 创建运行脚本的批处理文件run_backup.bat：

@echo off set PYTHONPATH=C:\Python38 call C:\Python38\python.exe C:\scripts\switch_backup.py

2. 使用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 $trigger

5.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: 'team@example.com' } catch (e) { emailext body: "备份失败: ${e}", subject: '备份失败警报', to: 'admin@example.com' error '备份失败' } } } } stage('Archive') { steps { archiveArtifacts artifacts: 'backups/**', fingerprint: true } } } }

5.3 备份文件生命周期管理

长期积累的备份文件会占用大量存储空间，需要制定保留策略：

def cleanup_old_backups(root_dir, keep_days=30): """自动清理超过指定天数的备份文件""" 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, topdown=False): 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', encoding='utf-8') as f: old_lines = f.readlines() with open(new_file, 'r', encoding='utf-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_workers=10): """使用线程池并行备份多台设备""" folder_path = create_folder() results = {} with ThreadPoolExecutor(max_workers=max_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_path

6.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 True

7. 安全增强措施

自动化备份系统需要特别注意安全性：

安全最佳实践：

1. 使用SSH密钥认证替代密码（如设备支持）
2. 配置最小权限账户进行备份
3. 加密存储设备凭证信息
4. 设置备份文件的访问权限控制
5. 定期轮换备份存储的加密密钥

凭证管理示例：

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)