我要评论本地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修改交换机内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论