交换机批量备份python脚本

批量自动化备份网络交换机的配置文件，支持Cisco、H3C和华为等主流厂商的设备。 记录python检查日志和备份结果。

运行环境

• python 3.8+、可以ssh目标交换机

如何运行程序

#文件同目录下创建 `switch_list.xlsx` 文件
#python switch_backup_check_latest.py

交换机清单文件

文件格式

必须包含以下列：

• IP: 交换机IP地址
• Username: SSH用户名
• Password: SSH密码
• Vendor: 设备厂商（cisco/h3c/huawei）
• EnablePassword (可选): 特权密码（用于进入特权模式）

文件样例

输出文件

备份文件

• 位置：./backups/ 目录
• 文件名格式：<IP>_<厂商>_<时间戳>.cfg
• 内容：交换机当前配置

备份日志

• 位置：./backups/backup_log.txt
• 内容：
  • 每次备份的记录（成功/失败）
  • 备份时间戳

运行日志

• 位置：当前目录下的 switch_backup.log
• 内容：
  • 程序运行详细日志
  • 设备连接状态
  • 错误和警告信息

代码

import pandas as pd  # 数据处理核心库，用于读取Excel设备清单和生成检查报告
import paramiko  # SSH客户端库，用于连接网络设备并执行命令
from datetime import datetime  # 日期时间处理，用于生成时间戳和报告文件名
import time  # 时间处理，用于SSH连接中的等待操作
import logging  # 日志记录，用于创建运行日志文件和控制台输出
from typing import Dict, List, Optional  # 类型注解，用于函数参数和返回值的类型提示
import os

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

# 备份配置
BACKUP_CONFIG = {
    'backup_dir': './backups',  # 备份文件存储目录
    'log_file': './backups/backup_log.txt',  # 备份日志文件
    'encoding': 'gbk',  # 华为设备默认编码
    'backup_commands': {
        'cisco': 'show running-config',
        'h3c': 'display current-configuration',
        'huawei': 'display current-configuration'
    },
    'timeout': 120  # 命令执行超时时间(秒)
}

def create_backup_directory():
    """创建备份目录和日志文件"""
    try:
        # 创建备份目录
        os.makedirs(BACKUP_CONFIG['backup_dir'], exist_ok=True)
        logger.info(f"备份目录已创建: {BACKUP_CONFIG['backup_dir']}")

        # 初始化日志文件
        if not os.path.exists(BACKUP_CONFIG['log_file']):
            with open(BACKUP_CONFIG['log_file'], 'w', encoding='utf-8') as f:
                f.write("交换机配置备份日志\n")
                f.write("="*50 + "\n")
                f.write(f"创建时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")

        return True
    except Exception as e:
        logger.error(f"创建备份目录失败: {str(e)}")
        return False

def backup_device(device: Dict[str, str]) -> bool:
    """
    备份单台交换机配置

    参数:
        device: 包含设备信息的字典

    返回:
        bool: 备份是否成功
    """
    ip = device['IP']
    vendor = device['Vendor'].lower()
    username = device['Username']
    password = device['Password']
    enable_pass = device.get('EnablePassword', '')  # 可选特权密码

    ssh = None
    try:
        logger.info(f"开始备份设备: {ip} ({vendor.upper()})")

        # 建立SSH连接
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(
            ip,
            username=username,
            password=password,
            timeout=15,
            banner_timeout=30,
            look_for_keys=False
        )

        # 获取交互式shell
        shell = ssh.invoke_shell()
        time.sleep(1)  # 等待连接稳定

        # 禁用分页
        if vendor == 'cisco':
            shell.send('terminal length 0\n')
        else:  # H3C/Huawei
            shell.send('screen-length disable\n')
        time.sleep(1)

        # 进入特权模式（如果需要）
        if enable_pass:
            shell.send('enable\n')
            time.sleep(0.5)
            shell.send(f'{enable_pass}\n')
            time.sleep(1)

        # 发送备份命令
        backup_cmd = BACKUP_CONFIG['backup_commands'].get(vendor, '')
        if not backup_cmd:
            logger.warning(f"{ip} 未找到厂商 {vendor} 的备份命令")
            return False

        shell.send(backup_cmd + '\n')
        time.sleep(1)  # 等待命令开始执行

        # 读取输出直到提示符出现
        output = ""
        while not output.endswith('#' if vendor == 'cisco' else ']'):  # 等待命令结束
            if shell.recv_ready():
                output += shell.recv(4096).decode(BACKUP_CONFIG['encoding'], errors='ignore')
            else:
                time.sleep(1)

        # 华为设备可能需要特殊编码处理
        if vendor == 'huawei':
            try:
                # 尝试用GBK解码
                output = output.encode('iso-8859-1').decode('gbk', errors='ignore')
            except:
                # 如果失败则使用UTF-8
                output = output.encode('iso-8859-1').decode('utf-8', errors='ignore')

        # 生成备份文件名
        timestamp = datetime.now().strftime("%Y%m%d_%H%M")
        filename = os.path.join(BACKUP_CONFIG['backup_dir'], f"{ip}_{vendor}_{timestamp}.cfg")

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

        # 记录备份日志
        with open(BACKUP_CONFIG['log_file'], 'a', encoding='utf-8') as log_file:
            log_file.write(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - {ip} ({vendor.upper()}) 备份成功\n")

        logger.info(f"{ip} 配置备份完成: {filename}")
        return True

    except paramiko.AuthenticationException:
        logger.error(f"{ip} 认证失败，请检查用户名/密码")
    except paramiko.SSHException as e:
        logger.error(f"{ip} SSH连接错误: {str(e)}")
    except Exception as e:
        logger.error(f"{ip} 备份失败: {str(e)}")
    finally:
        if ssh:
            ssh.close()

    # 记录失败日志
    with open(BACKUP_CONFIG['log_file'], 'a', encoding='utf-8') as log_file:
        log_file.write(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - {ip} ({vendor.upper()}) 备份失败\n")

    return False

def main():
    """主函数：读取设备清单并执行批量备份"""
    # 创建备份目录
    if not create_backup_directory():
        logger.error("无法创建备份目录，程序终止")
        return

    try:
        # 读取设备清单
        device_file = 'switch_list.xlsx'
        if not os.path.exists(device_file):
            logger.error(f"设备清单文件不存在: {device_file}")
            return

        df_devices = pd.read_excel(device_file)
        logger.info(f"成功读取 {len(df_devices)} 台设备信息")

        # 检查必要列
        required_columns = ['IP', 'Username', 'Password', 'Vendor']
        if not all(col in df_devices.columns for col in required_columns):
            missing = set(required_columns) - set(df_devices.columns)
            logger.error(f"缺少必要列: {', '.join(missing)}")
            return

        # 执行备份
        success_count = 0
        devices = df_devices.to_dict('records')

        for device in devices:
            if backup_device(device):
                success_count += 1

        # 输出备份摘要
        logger.info("\n备份完成! 备份摘要:")
        logger.info(f"- 总设备数: {len(devices)}")
        logger.info(f"- 成功备份: {success_count}")
        logger.info(f"- 失败设备: {len(devices) - success_count}")
        logger.info(f"- 备份目录: {os.path.abspath(BACKUP_CONFIG['backup_dir'])}")
        logger.info(f"- 备份日志: {os.path.abspath(BACKUP_CONFIG['log_file'])}")

        # 输出失败设备列表
        if success_count < len(devices):
            failed_ips = [d['IP'] for d in devices if not backup_device(d)]
            logger.warning(f"备份失败设备: {', '.join(failed_ips)}")

    except Exception as e:
        logger.error(f"主程序异常: {str(e)}", exc_info=True)

if __name__ == "__main__":
    main()

注意事项

• 确保交换机已启用SSH服务
• 网络可达目标交换机
• 使用的SSH账号有足够的权限执行检查命令，需要进入特权模式的要保留特权密码

# Linux cron示例（每周一凌晨2点备份）
0 2 * * 1 /usr/bin/python3 /path/switch_backup_check_latest.py

venv环境：

可使用虚拟运行环境隔离版本和依赖

#cd path\to\your\project                #进入项目文件夹
#python -m venv venv                    #生成venv虚拟环境
#.\venv\Scripts\activate                #cmd命令行下激活venv环境
#pip install pandas paramiko openpyxl   #为临时环境安装依赖包
#python switch_backup_check_latest.py   #虚拟VENV环境下运行
#deactivate                             #退出虚拟环境