解放双手!用Python自动化实现50台华为交换机配置备份

一、为什么需要自动化备份？

如果你是一名网络工程师，正在管理几十台甚至上百台华为交换机，那么你一定对这样的场景不陌生：每到月底要做配置备份，就得一台一台地登录设备，一遍又一遍地输入display current-configuration，再把输出复制粘贴到本地文件。这种“人肉运维”模式，在网络规模小的时候还能应付，一旦设备数量上来，效率低下不说，还容易因为手误敲错命令引发故障。

传统手动备份的痛点非常明显：

• 时间成本高：按每台设备5分钟计算，50台设备需要4个多小时
• 人为错误多：手工操作易出现命令错误、漏备份等问题
• 版本管理难：备份文件命名不规范，难以追踪历史变更

Python自动化方案能把备份时间从小时级缩短到分钟级，彻底解放你的双手。

二、技术选型：Paramiko vs Netmiko

实现SSH连接华为交换机，主要有两种Python库可选：

Paramiko 是Python的基础SSH库，功能强大但需要手动处理很多底层细节，比如交互式shell的建立、命令输出的读取等。

Netmiko 则是专门为网络设备自动化而生的库，它封装了SSH连接、会话保持、命令执行、输出解析等一系列繁琐的底层操作。对于华为交换机，Netmiko已经内置了完善的驱动支持，代码更简洁、更稳定。

对于50台交换机的批量备份场景，强烈推荐使用Netmiko。本文的示例代码也将基于Netmiko来实现。

三、环境准备

3.1 安装Python和Netmiko

首先需要一台安装了Python的电脑，Windows、macOS或Linux都可以。建议使用Python 3.7或更高版本。

安装Netmiko库：

pip install netmiko

如果安装速度慢，可以使用国内镜像源：

pip install netmiko -i https://pypi.tuna.tsinghua.edu.cn/simple

3.2 交换机端准备

在开始编写脚本之前，需要确保交换机满足以下条件：

• 管理电脑与交换机网络可达（能ping通）
• 交换机已开启SSH服务（华为交换机上用 stelnet server enable 命令开启）
• 拥有一个具有足够权限的SSH账号（至少要有display权限）

安全建议：单独为自动化脚本创建一个专用账号，并赋予固定权限，避免使用管理员账号。

3.3 准备设备清单

创建一个 devices.txt 文件，每行一个IP地址：

192.168.1.101
192.168.1.102
192.168.1.103
...（共50台）

四、单台交换机备份——基础篇

先从连接单台交换机开始，这是批量备份的基础。

from netmiko import ConnectHandler
import datetime
import os

# 交换机连接参数
device = {
'device_type': 'huawei',
'host': '192.168.1.101',
'username': 'admin',
'password': 'Admin@123',
'port': 22,
'timeout': 60,          # 连接超时时间
'session_timeout': 60,  # 会话超时时间
}

defbackup_single_device(device_params):
"""备份单台华为交换机配置"""
try:
# 建立SSH连接
        connection = ConnectHandler(**device_params)

# 关闭分页显示，确保能获取完整配置
        connection.send_command('screen-length 0 temporary')

# 执行备份命令
        output = connection.send_command('display current-configuration')

# 生成带时间戳的文件名
        timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
        filename = f"backup_{device_params['host']}_{timestamp}.txt"

# 保存到本地
with open(filename, 'w', encoding='utf-8') as f:
            f.write(output)

        connection.disconnect()
        print(f"✅ {device_params['host']} 备份成功: {filename}")
returnTrue

except Exception as e:
        print(f"❌ {device_params['host']} 备份失败: {e}")
returnFalse

# 执行备份
if __name__ == '__main__':
    backup_single_device(device)

五、批量备份50台交换机——进阶篇

5.1 从文件读取设备列表

defread_devices_from_file(file_path):
"""从文本文件读取设备IP列表"""
    devices = []
with open(file_path, 'r') as f:
for line in f:
            ip = line.strip()
if ip andnot ip.startswith('#'):  # 跳过空行和注释行
                devices.append({
'device_type': 'huawei',
'host': ip,
'username': 'admin',
'password': 'Admin@123',
'port': 22,
'timeout': 60,
'session_timeout': 60,
                })
return devices

5.2 批量备份（串行方式）

defbatch_backup_serial(devices, backup_dir='backups'):
"""串行批量备份（适合设备数量较少时）"""
# 创建备份目录
ifnot os.path.exists(backup_dir):
        os.makedirs(backup_dir)

    success_count = 0
    fail_count = 0
    fail_list = []

for device in devices:
        host = device['host']
try:
            connection = ConnectHandler(**device)
            connection.send_command('screen-length 0 temporary')
            output = connection.send_command('display current-configuration')

            timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
            filename = os.path.join(backup_dir, f"{host}_{timestamp}.txt")

with open(filename, 'w', encoding='utf-8') as f:
                f.write(output)

            connection.disconnect()
            print(f"✅ {host} 备份成功")
            success_count += 1

except Exception as e:
            print(f"❌ {host} 备份失败: {e}")
            fail_count += 1
            fail_list.append(host)

# 输出统计结果
    print(f"\n📊 备份完成！成功: {success_count}台，失败: {fail_count}台")
if fail_list:
        print(f"失败设备: {fail_list}")

return success_count, fail_count, fail_list

注意：串行方式备份50台设备，每台假设需要30秒，总共需要约25分钟。如果对时间有更高要求，可以考虑并发方式。

六、并发批量备份——高级篇

对于50台交换机，使用并发可以大幅提升效率。但需要特别注意：并发数不宜过大，否则可能对网络和设备造成压力。建议并发数控制在5-10之间。

6.1 使用线程池实现并发备份

from concurrent.futures import ThreadPoolExecutor, as_completed
import threading

# 线程安全的计数器
classBackupCounter:
def__init__(self):
        self.success = 0
        self.fail = 0
        self.lock = threading.Lock()
        self.fail_list = []

defadd_success(self):
with self.lock:
            self.success += 1

defadd_fail(self, host):
with self.lock:
            self.fail += 1
            self.fail_list.append(host)

defbackup_one_device(device, backup_dir, counter):
"""备份单台设备（供线程池调用）"""
    host = device['host']
try:
        connection = ConnectHandler(**device)
        connection.send_command('screen-length 0 temporary')
        output = connection.send_command('display current-configuration')

        timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
        filename = os.path.join(backup_dir, f"{host}_{timestamp}.txt")

with open(filename, 'w', encoding='utf-8') as f:
            f.write(output)

        connection.disconnect()
        print(f"✅ {host} 备份成功")
        counter.add_success()
returnTrue

except Exception as e:
        print(f"❌ {host} 备份失败: {e}")
        counter.add_fail(host)
returnFalse

defbatch_backup_concurrent(devices, backup_dir='backups', max_workers=5):
"""并发批量备份"""
# 创建备份目录
ifnot os.path.exists(backup_dir):
        os.makedirs(backup_dir)

    counter = BackupCounter()

# 使用线程池并发执行
with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(backup_one_device, device, backup_dir, counter): device
for device in devices
        }

# 等待所有任务完成
for future in as_completed(futures):
try:
                future.result()
except Exception as e:
                device = futures[future]
                print(f"⚠️ {device['host']} 任务异常: {e}")

# 输出统计结果
    print(f"\n📊 并发备份完成！成功: {counter.success}台，失败: {counter.fail}台")
if counter.fail_list:
        print(f"失败设备: {counter.fail_list}")

return counter.success, counter.fail, counter.fail_list

6.2 并发数的选择

• max_workers=5：适合50台设备的场景，既提升了效率，又不会对网络造成过大压力
• 备份50台设备，并发数为5时，理论上只需要串行时间的1/5，约5-8分钟即可完成

七、完整脚本示例

将以上模块整合成一个完整的脚本：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
华为交换机配置自动备份脚本
支持批量读取设备列表、并发备份、日志记录
"""

from netmiko import ConnectHandler
import datetime
import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('backup.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# ============ 配置区 ============
BACKUP_DIR = 'backups'# 备份文件存放目录
DEVICE_FILE = 'devices.txt'# 设备IP列表文件
MAX_WORKERS = 5# 并发数
USERNAME = 'admin'# SSH用户名
PASSWORD = 'Admin@123'# SSH密码
# ================================

classBackupCounter:
def__init__(self):
        self.success = 0
        self.fail = 0
        self.lock = threading.Lock()
        self.fail_list = []

defadd_success(self):
with self.lock:
            self.success += 1

defadd_fail(self, host):
with self.lock:
            self.fail += 1
            self.fail_list.append(host)

defread_devices(file_path):
"""读取设备列表"""
    devices = []
ifnot os.path.exists(file_path):
        logger.error(f"设备文件不存在: {file_path}")
return devices

with open(file_path, 'r') as f:
for line in f:
            ip = line.strip()
if ip andnot ip.startswith('#'):
                devices.append({
'device_type': 'huawei',
'host': ip,
'username': USERNAME,
'password': PASSWORD,
'port': 22,
'timeout': 60,
'session_timeout': 60,
                })

    logger.info(f"读取到 {len(devices)} 台设备")
return devices

defbackup_device(device, backup_dir, counter):
"""备份单台设备"""
    host = device['host']
try:
        logger.info(f"开始备份 {host}")
        connection = ConnectHandler(**device)
        connection.send_command('screen-length 0 temporary')
        output = connection.send_command('display current-configuration')

        timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
        filename = os.path.join(backup_dir, f"{host}_{timestamp}.txt")

with open(filename, 'w', encoding='utf-8') as f:
            f.write(output)

        connection.disconnect()
        logger.info(f"✅ {host} 备份成功")
        counter.add_success()
returnTrue

except Exception as e:
        logger.error(f"❌ {host} 备份失败: {e}")
        counter.add_fail(host)
returnFalse

defmain():
"""主函数"""
# 创建备份目录
ifnot os.path.exists(BACKUP_DIR):
        os.makedirs(BACKUP_DIR)
        logger.info(f"创建备份目录: {BACKUP_DIR}")

# 读取设备列表
    devices = read_devices(DEVICE_FILE)
ifnot devices:
        logger.error("没有可用的设备，程序退出")
        sys.exit(1)

    logger.info(f"开始批量备份，共 {len(devices)} 台设备，并发数: {MAX_WORKERS}")
    start_time = datetime.datetime.now()

    counter = BackupCounter()

# 并发执行备份
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {
            executor.submit(backup_device, device, BACKUP_DIR, counter): device
for device in devices
        }
for future in as_completed(futures):
try:
                future.result()
except Exception as e:
                device = futures[future]
                logger.error(f"⚠️ {device['host']} 任务异常: {e}")

# 统计结果
    elapsed = (datetime.datetime.now() - start_time).total_seconds()
    logger.info(f"\n{'='*50}")
    logger.info(f"📊 备份完成！")
    logger.info(f"   总设备数: {len(devices)} 台")
    logger.info(f"   成功: {counter.success} 台")
    logger.info(f"   失败: {counter.fail} 台")
if counter.fail_list:
        logger.info(f"   失败设备: {', '.join(counter.fail_list)}")
    logger.info(f"   耗时: {elapsed:.2f} 秒")
    logger.info(f"{'='*50}")

if __name__ == '__main__':
    main()

八、部署为定时任务

备份应该定期执行，而不是只在需要时手动运行。

Windows系统——任务计划程序

1. 打开“任务计划程序”
2. 创建基本任务，设置触发器为每天凌晨1点
3. 操作选择“启动程序”，填入：
• 程序：python.exe
• 参数：脚本的完整路径

Linux系统——Cron定时任务

在终端执行 crontab -e，添加以下行：

# 每天凌晨1点执行备份
0 1 * * * /usr/bin/python3 /path/to/backup_script.py >> /var/log/backup.log 2>&1

九、注意事项与最佳实践

9.1 安全建议

• 不要将密码明文写在脚本中：建议使用环境变量或加密的配置文件存储敏感信息
• 使用专用账号：为自动化脚本创建独立的SSH账号，权限最小化
• 定期更换密码：配合企业的密码管理策略

9.2 网络与性能

• 控制并发数：建议并发数不超过10，避免对交换机CPU造成过大压力
• 设置合理的超时时间：Netmiko默认连接超时只有10秒，建议调整为60秒以上
• 先测试后上线：建议先用1-2台测试交换机验证脚本，确认无误后再批量执行

9.3 备份文件管理

• 建议使用版本控制：将备份目录纳入Git管理，便于追踪配置变更历史
• 定期清理旧备份：避免备份文件无限增长占用磁盘空间
• 备份文件命名规范：建议格式为 {IP}_{日期时间}.txt，便于检索

十、总结

通过Python + Netmiko的组合，50台华为交换机的配置备份可以从4个多小时的手工劳动，缩短到几分钟的自动化执行。本文提供的脚本已经包含了从设备读取、并发备份到日志记录和错误处理的完整功能，你可以直接复制使用，并根据实际环境调整配置参数。

自动化运维的价值不仅在于节省时间，更在于消除人为失误、实现标准化操作和建立可追溯的配置变更历史。当你把双手从重复的CLI操作中解放出来，就能把更多精力投入到更有价值的网络设计和优化工作中去。