快捷導(dǎo)航

Python腳本實(shí)現(xiàn)批量修改交換機(jī)

更新時(shí)間：2026年05月11日 09:29:03 作者：normanhere

本文介紹了一個(gè)基于Python和Netmiko庫的多廠商交換機(jī)批量配置腳本,該腳本主要功能包括支持華三、華為、思科、銳捷、中興等主流品牌交換機(jī)配置,感興趣的小伙伴可以了解下

本地wimdows環(huán)境，安裝好python最新版本的，安裝netmiko庫。

所有文件位于C:\1 文件夾

編寫如下腳本命名為 batch_config.py 實(shí)現(xiàn)不同品牌 華三華為思科銳捷中興交換機(jī)的批量設(shè)置。

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. 品牌 → 命令文件映射 ----------
# 每個(gè)品牌可以有多個(gè)別名（如中文、英文簡(jiǎn)稱）
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'   # 未匹配品牌時(shí)使用的備用文件

# ---------- 3. 按品牌加載命令 ----------
def load_commands(vendor: str) -> list:
    """
    根據(jù)廠商品牌返回對(duì)應(yīng)的命令列表。
    優(yōu)先讀取品牌專屬文件，如果文件不存在或未配置映射則回退到默認(rèn)文件。
    """
    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} 不存在，嘗試使用默認(rèn)文件")
        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} 內(nèi)容為空")
        return commands
    except Exception as e:
        logger.error(f"讀取命令文件 {filepath} 失敗: {e}")
        return []

# ---------- 4. 單臺(tái)設(shè)備配置函數(shù) ----------
def configure_device(ip, username, password, vendor, secret, results):
    # 根據(jù)廠商選擇 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

    # ---------- 關(guān)鍵：按品牌加載專屬命令 ----------
    commands = load_commands(vendor)
    if not commands:
        logger.error(f"{ip} ({vendor}): 配置命令為空，跳過該設(shè)備")
        results['failed'].append((ip, '無可用配置命令'))
        return

    # 構(gòu)建設(shè)備連接參數(shù)
    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 密碼，則進(jìn)入特權(quán)模式
            if secret:
                conn.enable()
            output = conn.send_config_set(commands, exit_config_mode=False)
            # 保存配置（Netmiko 會(huì)根據(jù) device_type 自動(dòng)選擇合適的命令，如 save、write memory）
            conn.save_config()
            logger.info(f"? {ip} ({vendor}) 配置成功并保存")
            results['success'].append(ip)
    except NetmikoTimeoutException:
        logger.error(f"? {ip} ({vendor}) 連接超時(shí)")
        results['failed'].append((ip, '連接超時(shí)'))
    except NetmikoAuthenticationException:
        logger.error(f"? {ip} ({vendor}) 認(rèn)證失敗")
        results['failed'].append((ip, '認(rèn)證失敗'))
    except Exception as e:
        logger.error(f"? {ip} ({vendor}) 配置異常: {e}")
        results['failed'].append((ip, str(e)))

# ---------- 5. 主函數(shù) ----------
def main():
    # 讀取設(shè)備列表（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(),  # 默認(rèn) H3C，其他品牌必須明確填寫
                    row.get('secret', '').strip()      # 無 enable 密碼時(shí)留空
                ))
    except Exception as e:
        logger.error(f"讀取設(shè)備列表失敗: {e}")
        return

    if not devices:
        logger.error("設(shè)備列表為空！")
        return

    logger.info(f"共加載 {len(devices)} 臺(tái)設(shè)備，開始批量配置...")

    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()

        # 控制并發(fā)數(shù)
        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'])} 臺(tái)，失敗: {len(results['failed'])} 臺(tái)")
    if results['failed']:
        logger.error("失敗設(shè)備列表：")
        for ip, reason in results['failed']:
            logger.error(f"  {ip} - {reason}")

if __name__ == '__main__':
    main()

如下腳本為批量導(dǎo)出交換機(jī)配置，并保存為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'   # 無法匹配品牌時(shí)的備用命令

# ---------- 3. 單臺(tái)設(shè)備配置收集函數(shù) ----------
def collect_one_device(ip, username, password, vendor, secret, results):
    # 解析設(shè)備類型
    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

    # 獲取該品牌對(duì)應(yīng)的配置查看命令
    cmd = SHOW_CONFIG_CMD.get(vendor_lower, DEFAULT_CMD)

    # 連接參數(shù)
    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',  # 可選：保存會(huì)話日志
    }
    if secret:
        device['secret'] = secret

    try:
        logger.info(f"正在連接 {ip} ({vendor}) 并獲取配置...")
        with ConnectHandler(**device) as conn:
            if secret:
                conn.enable()      # 進(jìn)入特權(quán)模式（如果需要）
            # 執(zhí)行命令，獲取完整配置輸出
            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}) 連接超時(shí)")
        results['failed'].append((ip, '連接超時(shí)'))
    except NetmikoAuthenticationException:
        logger.error(f"? {ip} ({vendor}) 認(rèn)證失敗")
        results['failed'].append((ip, '認(rèn)證失敗'))
    except Exception as e:
        logger.error(f"? {ip} ({vendor}) 異常: {e}")
        results['failed'].append((ip, str(e)))

# ---------- 4. 主函數(shù) ----------
def main():
    # 讀取設(shè)備列表（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"讀取設(shè)備列表失敗: {e}")
        return

    if not devices:
        logger.error("設(shè)備列表為空！")
        return

    logger.info(f"共加載 {len(devices)} 臺(tái)設(shè)備，開始批量收集配置...")

    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'])} 臺(tái)，失敗: {len(results['failed'])} 臺(tái)")
    if results['failed']:
        logger.error("失敗設(shè)備列表：")
        for ip, reason in results['failed']:
            logger.error(f"  {ip} - {reason}")

if __name__ == '__main__':
    main()

最終目錄結(jié)構(gòu)如下：

devices.csv結(jié)構(gòu)如下：

實(shí)現(xiàn)的最終效果是：

運(yùn)行batch_config.py 并發(fā)（10并發(fā)默認(rèn)）讀取devices.csv 按照設(shè)備品牌判斷，調(diào)用對(duì)應(yīng)腳本配置，并自動(dòng)保存配置過程日志；輸出配置成功多少臺(tái)，配置錯(cuò)誤多少臺(tái)。

運(yùn)行collect_switch_configs.py 并發(fā)讀?。?0并發(fā)默認(rèn)）讀取devices.csv 將配置文件按照管理IP地址命名保存到configs文件夾。

到此這篇關(guān)于Python腳本實(shí)現(xiàn)批量修改交換機(jī)的文章就介紹到這了,更多相關(guān)Python修改交換機(jī)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: