Python脚本实现批量修改交换机
更新时间：2026年05月11日 09:29:03 作者：normanhere
本文介绍了一个基于Python和Netmiko库的多厂商交换机批量配置脚本,该脚本主要功能包括支持华三、华为、思科、锐捷、中兴等主流品牌交换机配置,感兴趣的小伙伴可以了解下
本地wimdows环境，安装好python最新版本的，安装netmiko库。
所有文件位于C:\1 文件夹
编写如下脚本命名为 batch_config.py 实现不同品牌 华三 华为 思科 锐捷 中兴交换机的批量设置。
import csv
import logging
import sys
import threading
import os
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
# ---------- 1. 日志配置 ----------
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(threadName)s - %(message)s',
handlers=[
logging.FileHandler(r'c:\1\batch_result.log', encoding='utf-8'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
# ---------- 2. 品牌 → 命令文件映射 ----------
# 每个品牌可以有多个别名（如中文、英文简称）
COMMAND_FILE_MAP = {
'h3c': r'c:\1\config_commands_h3c.txt',
'ruijie': r'c:\1\config_commands_ruijie.txt',
'锐捷': r'c:\1\config_commands_ruijie.txt',
'zte': r'c:\1\config_commands_zte.txt',
'中兴': r'c:\1\config_commands_zte.txt',
'huawei': r'c:\1\config_commands_huawei.txt',
'华为': r'c:\1\config_commands_huawei.txt',
'cisco': r'c:\1\config_commands_cisco.txt',
'思科': r'c:\1\config_commands_cisco.txt',
}
DEFAULT_COMMAND_FILE = r'c:\1\config_commands.txt' # 未匹配品牌时使用的备用文件
# ---------- 3. 按品牌加载命令 ----------
def load_commands(vendor: str) -> list:
"""
根据厂商品牌返回对应的命令列表。
优先读取品牌专属文件，如果文件不存在或未配置映射则回退到默认文件。
"""
vendor_key = vendor.strip().lower()
filepath = COMMAND_FILE_MAP.get(vendor_key, DEFAULT_COMMAND_FILE)
if not os.path.exists(filepath):
logger.warning(f"{vendor} 的命令文件 {filepath} 不存在，尝试使用默认文件")
filepath = DEFAULT_COMMAND_FILE
try:
with open(filepath, 'r', encoding='utf-8') as f:
commands = [line.strip() for line in f
if line.strip() and not line.strip().startswith('#')]
if not commands:
logger.warning(f"命令文件 {filepath} 内容为空")
return commands
except Exception as e:
logger.error(f"读取命令文件 {filepath} 失败: {e}")
return []
# ---------- 4. 单台设备配置函数 ----------
def configure_device(ip, username, password, vendor, secret, results):
# 根据厂商选择 device_type
vendor_lower = vendor.strip().lower()
if vendor_lower == 'h3c':
device_type = 'hp_comware'
elif vendor_lower in ('ruijie', '锐捷'):
device_type = 'ruijie_os'
elif vendor_lower in ('zte', '中兴'):
device_type = 'zte_zxros'
elif vendor_lower in ('huawei', '华为'):
device_type = 'huawei'
elif vendor_lower in ('cisco', '思科'):
device_type = 'cisco_ios'
else:
logger.error(f"{ip}: 不支持的厂商 {vendor}")
results['failed'].append((ip, f'不支持的厂商: {vendor}'))
return
# ---------- 关键：按品牌加载专属命令 ----------
commands = load_commands(vendor)
if not commands:
logger.error(f"{ip} ({vendor}): 配置命令为空，跳过该设备")
results['failed'].append((ip, '无可用配置命令'))
return
# 构建设备连接参数
device = {
'device_type': device_type,
'ip': ip,
'username': username,
'password': password,
'timeout': 30,
'session_timeout': 60,
'session_log': f'c:\\1\\{ip}_session.log',
}
if secret:
device['secret'] = secret
try:
logger.info(f"正在连接 {ip} ({vendor})...")
with ConnectHandler(**device) as conn:
# 如果配置了 enable 密码，则进入特权模式
if secret:
conn.enable()
output = conn.send_config_set(commands, exit_config_mode=False)
# 保存配置（Netmiko 会根据 device_type 自动选择合适的命令，如 save、write memory）
conn.save_config()
logger.info(f"✅ {ip} ({vendor}) 配置成功并保存")
results['success'].append(ip)
except NetmikoTimeoutException:
logger.error(f"❌ {ip} ({vendor}) 连接超时")
results['failed'].append((ip, '连接超时'))
except NetmikoAuthenticationException:
logger.error(f"❌ {ip} ({vendor}) 认证失败")
results['failed'].append((ip, '认证失败'))
except Exception as e:
logger.error(f"❌ {ip} ({vendor}) 配置异常: {e}")
results['failed'].append((ip, str(e)))
# ---------- 5. 主函数 ----------
def main():
# 读取设备列表（CSV表头：ip, username, password, vendor, secret）
devices = []
try:
with open(r'c:\1\devices.csv', 'r', encoding='utf-8-sig') as f:
reader = csv.DictReader(f)
for row in reader:
devices.append((
row['ip'].strip(),
row['username'].strip(),
row['password'].strip(),
row.get('vendor', 'h3c').strip(), # 默认 H3C，其他品牌必须明确填写
row.get('secret', '').strip() # 无 enable 密码时留空
))
except Exception as e:
logger.error(f"读取设备列表失败: {e}")
return
if not devices:
logger.error("设备列表为空！")
return
logger.info(f"共加载 {len(devices)} 台设备，开始批量配置...")
results = {'success': [], 'failed': []}
threads = []
MAX_THREADS = 10
for ip, user, pwd, vendor, secret in devices:
t = threading.Thread(
target=configure_device,
args=(ip, user, pwd, vendor, secret, results),
name=f"Thread-{ip}"
)
threads.append(t)
t.start()
# 控制并发数
if len(threads) >= MAX_THREADS:
for t in threads:
t.join()
threads = []
for t in threads:
t.join()
logger.info("=" * 50)
logger.info(f"配置完成！成功: {len(results['success'])} 台，失败: {len(results['failed'])} 台")
if results['failed']:
logger.error("失败设备列表：")
for ip, reason in results['failed']:
logger.error(f" {ip} - {reason}")
if __name__ == '__main__':
main()
如下脚本为批量导出交换机配置，并保存为collect_switch_configs.py 所有文件位于C:\1 文件夹
import csv
import logging
import sys
import threading
import os
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
# ---------- 1. 日志配置 ----------
LOG_FILE = r'c:\1\collect_configs.log'
OUTPUT_DIR = r'c:\1\configs' # 配置保存目录
os.makedirs(OUTPUT_DIR, exist_ok=True)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(threadName)s - %(message)s',
handlers=[
logging.FileHandler(LOG_FILE, encoding='utf-8'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
# ---------- 2. 品牌 → 配置查看命令映射 ----------
SHOW_CONFIG_CMD = {
'h3c': 'display current-configuration',
'ruijie': 'show running-config',
'锐捷': 'show running-config',
'zte': 'show running-config',
'中兴': 'show running-config',
'huawei': 'display current-configuration',
'华为': 'display current-configuration',
'cisco': 'show running-config',
'思科': 'show running-config',
}
DEFAULT_CMD = 'show running-config' # 无法匹配品牌时的备用命令
# ---------- 3. 单台设备配置收集函数 ----------
def collect_one_device(ip, username, password, vendor, secret, results):
# 解析设备类型
vendor_lower = vendor.strip().lower()
if vendor_lower == 'h3c':
device_type = 'hp_comware'
elif vendor_lower in ('ruijie', '锐捷'):
device_type = 'ruijie_os'
elif vendor_lower in ('zte', '中兴'):
device_type = 'zte_zxros'
elif vendor_lower in ('huawei', '华为'):
device_type = 'huawei'
elif vendor_lower in ('cisco', '思科'):
device_type = 'cisco_ios'
else:
logger.error(f"{ip}: 不支持的厂商 {vendor}")
results['failed'].append((ip, f'不支持的厂商: {vendor}'))
return
# 获取该品牌对应的配置查看命令
cmd = SHOW_CONFIG_CMD.get(vendor_lower, DEFAULT_CMD)
# 连接参数
device = {
'device_type': device_type,
'ip': ip,
'username': username,
'password': password,
'timeout': 30,
'session_timeout': 60,
'session_log': f'c:\\1\\{ip}_collect_session.log', # 可选：保存会话日志
}
if secret:
device['secret'] = secret
try:
logger.info(f"正在连接 {ip} ({vendor}) 并获取配置...")
with ConnectHandler(**device) as conn:
if secret:
conn.enable() # 进入特权模式（如果需要）
# 执行命令，获取完整配置输出
config_output = conn.send_command(cmd, read_timeout=60)
# 保存到本地文件
filename = os.path.join(OUTPUT_DIR, f"{ip}_config.txt")
with open(filename, 'w', encoding='utf-8') as f:
f.write(config_output)
logger.info(f"✅ {ip} ({vendor}) 配置已保存 -> {filename}")
results['success'].append(ip)
except NetmikoTimeoutException:
logger.error(f"❌ {ip} ({vendor}) 连接超时")
results['failed'].append((ip, '连接超时'))
except NetmikoAuthenticationException:
logger.error(f"❌ {ip} ({vendor}) 认证失败")
results['failed'].append((ip, '认证失败'))
except Exception as e:
logger.error(f"❌ {ip} ({vendor}) 异常: {e}")
results['failed'].append((ip, str(e)))
# ---------- 4. 主函数 ----------
def main():
# 读取设备列表（CSV表头：ip, username, password, vendor, secret）
devices = []
try:
with open(r'c:\1\devices.csv', 'r', encoding='utf-8-sig') as f:
reader = csv.DictReader(f)
for row in reader:
devices.append((
row['ip'].strip(),
row['username'].strip(),
row['password'].strip(),
row.get('vendor', 'h3c').strip(),
row.get('secret', '').strip()
))
except Exception as e:
logger.error(f"读取设备列表失败: {e}")
return
if not devices:
logger.error("设备列表为空！")
return
logger.info(f"共加载 {len(devices)} 台设备，开始批量收集配置...")
results = {'success': [], 'failed': []}
threads = []
MAX_THREADS = 10
for ip, user, pwd, vendor, secret in devices:
t = threading.Thread(
target=collect_one_device,
args=(ip, user, pwd, vendor, secret, results),
name=f"Thread-{ip}"
)
threads.append(t)
t.start()
if len(threads) >= MAX_THREADS:
for t in threads:
t.join()
threads = []
for t in threads:
t.join()
# 输出汇总
logger.info("=" * 50)
logger.info(f"收集完成！成功: {len(results['success'])} 台，失败: {len(results['failed'])} 台")
if results['failed']:
logger.error("失败设备列表：")
for ip, reason in results['failed']:
logger.error(f" {ip} - {reason}")
if __name__ == '__main__':
main()
最终目录结构如下：
devices.csv结构如下：
实现的最终效果是：
运行batch_config.py 并发（10并发默认）读取devices.csv 按照设备品牌判断，调用对应脚本配置，并自动保存配置过程日志；输出配置成功多少台，配置错误多少台。
运行collect_switch_configs.py 并发读取（10并发默认）读取devices.csv 将配置文件按照管理IP地址命名 保存到configs文件夹。
到此这篇关于Python脚本实现批量修改交换机的文章就介绍到这了,更多相关Python修改交换机内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
随着人工智能的不断发展，机器学习这门技术也越来越重要，很多人都开启了学习机器学习，本文就介绍了机器学习的基础内容，了解python爬虫，本文给大家分享Python爬取贝壳房源数据的实战教程，感兴趣的朋友一起学习吧2021-05-05
这篇文章主要介绍了keras中的backend.clip用法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-05-05
这篇文章主要介绍了关于Python中模块的简介、定义与使用,模块是Python的重要组成部分,需要的朋友可以参考下2023-04-04
这篇文章主要给大家介绍了关于OpenCV简单标准数字识别的相关资料,要通过opencv 进行数字识别离不开训练库的支持，需要对目标图片进行大量的训练，才能做到精准的识别出目标数字,需要的朋友可以参考下2021-09-09
今天小编就为大家分享一篇pytorch 使用加载训练好的模型做inference，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-02-02
python使用wmi模块获取windows下硬盘信息的方法
这篇文章主要介绍了python使用wmi模块获取windows下硬盘信息的方法,涉及Python获取系统硬件信息的相关技巧,需要的朋友可以参考下2015-05-05
随着短视频的大火，不仅可以给人们带来娱乐，还有热点新闻时事以及各种知识，刷短视频也逐渐成为了日常生活的一部分。本文将通过Pyhton依托Selenium来爬取短视频热点，需要的可以参考一下2022-04-04
Python中操作MySQL和SQL Server数据库的实战指南
本文详细介绍了如何在Pvthon中使用pymysql和pymssql库进行MySQL和SQLServer数据库的基本操作,包括连接、查询、插入、更新、异常处理、参数化查询、ORM应用、事务处理以及连接池的使用,为读者提供了一套全面的数据库操作指南,需要的朋友可以参考下2025-11-11
这篇文章主要介绍了python 如何调用 dubbo 接口，帮助大家更好的理解和学习python，感兴趣的朋友可以了解下2020-09-09
这篇文章主要介绍了Numpy中np.max的用法及np.maximum区别，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2020-11-11
最新评论