第326篇：多厂商设备统一管理框架

关键词

多厂商管理、设备抽象、统一接口、适配器模式、厂商适配、设备驱动、跨厂商自动化

一、为什么需要统一管理

1.1 多厂商网络的现实

现实中的多厂商网络：

  企业在网设备通常来自多个厂商：
  ┌──────────────────────────────────────────┐
  │  核心层                                  │
  │  ┌─ Cisco Nexus 9000 (DC 核心)          │
  │  ├─ Huawei NE系列 (骨干)                │
  │  └─ Juniper MX (边缘)                   │
  │                                           │
  │  汇聚层                                  │
  │  ┌─ Huawei S12700 (园区汇聚)            │
  │  └─ Arista 7280 (DC 接入)              │
  │                                           │
  │  接入层                                  │
  │  ┌─ Huawei S5700 (办公接入)             │
  │  ├─ H3C S5560 (生产网络)               │
  │  └─ Cisco Catalyst 2960 (旧区)          │
  └──────────────────────────────────────────┘

  每个厂商的差异：
  ┌─ 命令风格不同：display vs show
  ├─ 配置语法不同：vlan batch vs vlan database
  ├─ 管理协议不同：SSH/Telnet/NETCONF/API
  ├─ 查询输出格式不同
  └─ 配置保存方式不同

  直接后果：
  ┌─ 脚本需要为每个厂商写不同的版本
  ├─ 维护成本随厂商数量指数增长
  ├─ 新人学习成本高（要会 N 种 CLI）
  └─ 统一变更变成"不可能的任务"

1.2 统一管理架构

多厂商统一管理架构：

二、统一设备抽象

2.1 抽象接口定义

#!/usr/bin/env python3
# device_abstraction.py — 设备抽象接口

from abc import ABC, ABCMeta, abstractmethod
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
from enum import Enum


# =============================================
# 通用数据类型
# =============================================

@dataclass
class InterfaceInfo:
    """接口信息"""
    name: str
    description: str = ""
    mac_address: str = ""
    ip_address: str = ""
    subnet_mask: str = ""
    admin_status: str = "up"
    oper_status: str = "up"
    speed: str = ""
    mtu: int = 1500
    interface_type: str = ""  # GigabitEthernet, 10GE, etc.
    vlan_id: Optional[int] = None
    errors: Dict = field(default_factory=dict)
    counters: Dict = field(default_factory=dict)


@dataclass
class BGPInfo:
    """BGP 信息"""
    local_as: int
    router_id: str
    peers: Dict[str, Dict]  # peer_ip -> state
    total_prefixes: int = 0


@dataclass
class RouteInfo:
    """路由信息"""
    destination: str
    nexthop: str
    protocol: str  # BGP, OSPF, Static, etc.
    preference: int
    metric: int
    interface: str
    age: str = ""


@dataclass
class DeviceFacts:
    """设备基本信息"""
    hostname: str
    vendor: str         # huawei, cisco, juniper, etc.
    model: str
    os_version: str
    serial_number: str
    uptime: str
    cpu_usage: float = 0.0
    memory_usage: float = 0.0


# =============================================
# 统一设备接口
# =============================================

class BaseDevice(ABC):
    """设备抽象基类 - 所有厂商适配器必须实现"""

    def __init__(self, connection_info: Dict):
        self.connection_info = connection_info
        self.hostname = ""
        self._connected = False

    @abstractmethod
    def connect(self) -> bool:
        """建立设备连接"""
        pass

    @abstractmethod
    def disconnect(self):
        """断开设备连接"""
        pass

    # ── 基本信息 ──

    @abstractmethod
    def get_facts(self) -> DeviceFacts:
        """获取设备基本信息"""
        pass

    # ── 接口 ──

    @abstractmethod
    def get_interfaces(self) -> Dict[str, InterfaceInfo]:
        """获取所有接口状态"""
        pass

    @abstractmethod
    def get_interface(self, name: str) -> Optional[InterfaceInfo]:
        """获取指定接口信息"""
        pass

    # ── BGP ──

    @abstractmethod
    def get_bgp_peers(self) -> List[Dict]:
        """获取 BGP 邻居信息"""
        pass

    # ── 路由 ──

    @abstractmethod
    def get_routing_table(self) -> List[RouteInfo]:
        """获取路由表"""
        pass

    # ── 配置 ──

    @abstractmethod
    def get_config(self) -> str:
        """获取当前配置"""
        pass

    @abstractmethod
    def send_commands(
        self, commands: List[str]
    ) -> List[str]:
        """发送配置命令"""
        pass

    @abstractmethod
    def save_config(self) -> bool:
        """保存配置"""
        pass

    # ── LLDP ──

    @abstractmethod
    def get_lldp_neighbors(self) -> Dict:
        """获取 LLDP 邻居"""
        pass

    # ── 连接测试 ──

    @abstractmethod
    def ping(self, target: str, count: int = 3) -> Dict:
        """Ping 测试"""
        pass

2.2 华为 VRP 适配器

#!/usr/bin/env python3
# huawei_adapter.py — 华为 VRP 设备适配器

from netmiko import ConnectHandler
from device_abstraction import (
    BaseDevice, InterfaceInfo, BGPInfo,
    RouteInfo, DeviceFacts,
)
from typing import List, Dict, Optional, Any
import re
import time


class HuaweiVRPDevice(BaseDevice):
    """华为 VRP 设备适配器"""

    def __init__(self, connection_info: Dict):
        super().__init__(connection_info)
        self._conn = None

    def connect(self) -> bool:
        try:
            self._conn = ConnectHandler(**self.connection_info)
            self._conn.enable()
            self.hostname = self._conn.find_prompt().rstrip("#>")
            self._connected = True
            return True
        except Exception as e:
            print(f"华为设备连接失败: {e}")
            return False

    def disconnect(self):
        if self._conn:
            self._conn.disconnect()
            self._connected = False

    def get_facts(self) -> DeviceFacts:
        version_output = self._conn.send_command("display version")
        device_output = self._conn.send_command(
            "display device"
        )

        # 解析版本信息
        model = ""
        os_ver = ""
        for line in version_output.splitlines():
            if "VRP" in line:
                model_match = re.search(r"(\S+) VRP", line)
                if model_match:
                    model = model_match.group(1)
                ver_match = re.search(
                    r"VRP\s+\(R\)\s+software.*Version\s+([\d.]+)",
                    line,
                )
                if ver_match:
                    os_ver = ver_match.group(1)

        # 解析序列号
        serial = ""
        for line in device_output.splitlines():
            if "SERIAL" in line:
                parts = line.split()
                if len(parts) >= 2:
                    serial = parts[-1]

        # 解析 uptime
        uptime = ""
        for line in version_output.splitlines():
            if "uptime" in line.lower():
                uptime = line.strip()

        return DeviceFacts(
            hostname=self.hostname,
            vendor="huawei",
            model=model,
            os_version=os_ver,
            serial_number=serial,
            uptime=uptime,
        )

    def get_interfaces(self) -> Dict[str, InterfaceInfo]:
        output = self._conn.send_command("display interface brief")
        interfaces = {}

        for line in output.splitlines():
            if not line.strip() or "Interface" in line:
                continue

            parts = line.split()
            if len(parts) < 4:
                continue

            name = parts[0]
            interfaces[name] = InterfaceInfo(
                name=name,
                admin_status=parts[1].lower(),
                oper_status=parts[2].lower(),
            )

        return interfaces

    def get_interface(
        self, name: str
    ) -> Optional[InterfaceInfo]:
        interfaces = self.get_interfaces()
        return interfaces.get(name)

    def get_bgp_peers(self) -> List[Dict]:
        output = self._conn.send_command("display bgp peer")
        peers = []

        for line in output.splitlines():
            if "Established" in line or "Idle" in line or "Active" in line:
                parts = line.split()
                if len(parts) >= 4:
                    peers.append({
                        "peer_ip": parts[0],
                        "as": parts[1],
                        "state": parts[2],
                        "uptime": parts[3] if len(parts) > 3 else "",
                    })

        return peers

    def get_routing_table(self) -> List[RouteInfo]:
        output = self._conn.send_command(
            "display ip routing-table"
        )
        routes = []

        for line in output.splitlines():
            # 华为路由表格式
            # Destination/Mask  Proto  Pre  Cost  NextHop  Interface
            if not line.strip() or "Destination" in line:
                continue

            parts = line.split()
            if len(parts) >= 6:
                routes.append(RouteInfo(
                    destination=parts[0],
                    protocol=parts[1],
                    preference=int(parts[2]),
                    metric=int(parts[3]),
                    nexthop=parts[4],
                    interface=parts[5],
                ))

        return routes

    def get_config(self) -> str:
        return self._conn.send_command(
            "display current-configuration"
        )

    def send_commands(
        self, commands: List[str]
    ) -> List[str]:
        results = []
        for cmd in commands:
            output = self._conn.send_config_set(
                [cmd], read_timeout=30
            )
            results.append(output)

            if "Error" in output:
                print(f"命令执行错误: {cmd} -> {output}")

        return results

    def save_config(self) -> bool:
        try:
            self._conn.save_config()
            return True
        except Exception:
            return False

    def get_lldp_neighbors(self) -> Dict:
        output = self._conn.send_command(
            "display lldp neighbor brief"
        )
        neighbors = {}

        for line in output.splitlines():
            if not line.strip() or "Local" in line:
                continue
            parts = line.split()
            if len(parts) >= 4:
                local_port = parts[0]
                neighbors[local_port] = {
                    "remote_device": parts[1],
                    "remote_port": parts[2],
                }

        return neighbors

    def ping(self, target: str, count: int = 3) -> Dict:
        output = self._conn.send_command(
            f"ping {target} -c {count}"
        )
        loss = "100%"
        rtt_min = rtt_avg = rtt_max = 0

        loss_match = re.search(
            r"(\d+)% packet loss", output
        )
        if loss_match:
            loss = loss_match.group(1) + "%"

        rtt_match = re.search(
            r"min/avg/max.*?= (\S+)", output
        )
        if rtt_match:
            rtt_values = rtt_match.group(1).split("/")
            if len(rtt_values) >= 3:
                rtt_min, rtt_avg, rtt_max = rtt_values[:3]

        return {
            "target": target,
            "packet_loss": loss,
            "rtt_min": rtt_min,
            "rtt_avg": rtt_avg,
            "rtt_max": rtt_max,
            "reachable": "100%" not in loss,
        }

2.3 Cisco IOS 适配器

#!/usr/bin/env python3
# cisco_adapter.py — 思科 IOS 设备适配器

from netmiko import ConnectHandler
from device_abstraction import (
    BaseDevice, InterfaceInfo, RouteInfo, DeviceFacts,
)
from typing import List, Dict, Optional
import re


class CiscoIOSDevice(BaseDevice):
    """思科 IOS 设备适配器"""

    def __init__(self, connection_info: Dict):
        super().__init__(connection_info)
        self._conn = None

    def connect(self) -> bool:
        try:
            self._conn = ConnectHandler(**self.connection_info)
            self._conn.enable()
            self.hostname = self._conn.find_prompt().rstrip("#>")
            self._connected = True
            return True
        except Exception as e:
            print(f"思科设备连接失败: {e}")
            return False

    def disconnect(self):
        if self._conn:
            self._conn.disconnect()
            self._connected = False

    def get_facts(self) -> DeviceFacts:
        output = self._conn.send_command("show version")

        model = ""
        os_ver = ""
        serial = ""
        uptime = ""

        for line in output.splitlines():
            if "model" in line.lower():
                parts = line.split(",")
                for p in parts:
                    if "model" in p.lower():
                        model = p.split()[-1]
            if "Version" in line and "Software" in line:
                ver_match = re.search(
                    r"Version\s+([\d.()a-zA-Z]+)", line
                )
                if ver_match:
                    os_ver = ver_match.group(1)
            if "Processor board ID" in line:
                serial = line.split()[-1]
            if "uptime" in line.lower():
                uptime = line.strip()

        return DeviceFacts(
            hostname=self.hostname,
            vendor="cisco",
            model=model,
            os_version=os_ver,
            serial_number=serial,
            uptime=uptime,
        )

    def get_interfaces(self) -> Dict[str, InterfaceInfo]:
        output = self._conn.send_command(
            "show ip interface brief"
        )
        interfaces = {}

        for line in output.splitlines():
            if "Interface" in line or not line.strip():
                continue

            parts = line.split()
            if len(parts) < 4:
                continue

            name = parts[0]
            interfaces[name] = InterfaceInfo(
                name=name,
                ip_address=parts[1],
                admin_status=parts[3].lower(),
                oper_status=parts[4].lower(),
            )

        return interfaces

    def get_bgp_peers(self) -> List[Dict]:
        output = self._conn.send_command("show bgp summary")
        peers = []

        for line in output.splitlines():
            # Neighbor  AS  State  Up/Down  Prefixes
            if not line.strip() or "Neighbor" in line:
                continue
            parts = line.split()
            if len(parts) >= 5:
                peers.append({
                    "peer_ip": parts[0],
                    "as": parts[1],
                    "state": parts[2],
                    "uptime": parts[3],
                    "prefixes": parts[4],
                })

        return peers

    def get_routing_table(self) -> List[RouteInfo]:
        output = self._conn.send_command(
            "show ip route"
        )
        routes = []

        for line in output.splitlines():
            # O   10.0.0.0/24 [110/10] via 192.168.1.1, 00:01:23, GigabitEthernet0/0
            route_match = re.match(
                r"([A-Z]+)\s+([\d./]+)\s+\[(\d+)/(\d+)\]\s+via\s+"
                r"([\d.]+),\s+(\S+),\s+(\S+)",
                line,
            )
            if route_match:
                routes.append(RouteInfo(
                    destination=route_match.group(2),
                    protocol=route_match.group(1),
                    preference=int(route_match.group(3)),
                    metric=int(route_match.group(4)),
                    nexthop=route_match.group(5),
                    interface=route_match.group(7),
                    age=route_match.group(6),
                ))

        return routes

    def get_config(self) -> str:
        return self._conn.send_command("show running-config")

    def send_commands(
        self, commands: List[str]
    ) -> List[str]:
        results = []
        for cmd in commands:
            output = self._conn.send_config_set(
                [cmd], read_timeout=30
            )
            results.append(output)
        return results

    def save_config(self) -> bool:
        try:
            self._conn.send_command("write memory")
            return True
        except Exception:
            return False

    def get_lldp_neighbors(self) -> Dict:
        output = self._conn.send_command(
            "show lldp neighbors"
        )
        neighbors = {}

        for line in output.splitlines():
            parts = line.split()
            if len(parts) >= 5 and parts[0].startswith("Gi"):
                neighbors[parts[0]] = {
                    "remote_device": parts[1],
                    "remote_port": parts[2],
                }

        return neighbors

    def ping(self, target: str, count: int = 3) -> Dict:
        output = self._conn.send_command(
            f"ping {target} repeat {count}"
        )
        return {
            "target": target,
            "reachable": "Success" in output,
            "raw_output": output[:200],
        }

三、统一设备管理器

3.1 工厂模式

#!/usr/bin/env python3
# device_manager.py — 统一设备管理器

from device_abstraction import BaseDevice
from huawei_adapter import HuaweiVRPDevice
from cisco_adapter import CiscoIOSDevice
from typing import Dict, Optional


class DeviceFactory:
    """设备工厂 — 根据厂商创建对应的适配器"""

    # 厂商适配器注册表
    _adapters = {
        "huawei": HuaweiVRPDevice,
        "huawei_vrp": HuaweiVRPDevice,
        "cisco": CiscoIOSDevice,
        "cisco_ios": CiscoIOSDevice,
        "cisco_xe": CiscoIOSDevice,
        # 可继续扩展
        # "juniper": JuniperDevice,
        # "h3c": H3CDevice,
        # "arista": AristaDevice,
    }

    @classmethod
    def register_adapter(cls, vendor: str, adapter_cls):
        """注册新的厂商适配器"""
        cls._adapters[vendor] = adapter_cls

    @classmethod
    def create_device(
        cls, connection_info: Dict
    ) -> Optional[BaseDevice]:
        """根据连接配置创建设备实例"""
        device_type = connection_info.get("device_type", "")
        vendor = connection_info.get("vendor", "")

        # 从 device_type 推断厂商
        if not vendor:
            for key in cls._adapters:
                if key in device_type:
                    vendor = key
                    break

        if not vendor:
            raise ValueError(
                f"不支持的设备类型: {device_type}，"
                f"支持的厂商: {list(cls._adapters.keys())}"
            )

        adapter_cls = cls._adapters.get(vendor)
        if adapter_cls:
            return adapter_cls(connection_info)

        raise ValueError(f"未知厂商: {vendor}")


class UnifiedDeviceManager:
    """统一设备管理器"""

    def __init__(self):
        self.devices: Dict[str, BaseDevice] = {}

    def add_device(
        self, device_id: str, connection_info: Dict
    ) -> bool:
        """添加并连接设备"""
        try:
            device = DeviceFactory.create_device(
                connection_info
            )
            if device.connect():
                self.devices[device_id] = device
                print(
                    f"设备 {device_id} "
                    f"({device.hostname}) 连接成功"
                )
                return True
            else:
                print(f"设备 {device_id} 连接失败")
                return False
        except Exception as e:
            print(f"添加设备 {device_id} 失败: {e}")
            return False

    def remove_device(self, device_id: str):
        """移除并断开设备"""
        if device_id in self.devices:
            self.devices[device_id].disconnect()
            del self.devices[device_id]

    def get_device(self, device_id: str) -> BaseDevice:
        """获取设备实例"""
        return self.devices.get(device_id)

    def execute_on_all(
        self, method: str, *args, **kwargs
    ) -> Dict:
        """在所有设备上执行同一方法"""
        results = {}
        for device_id, device in self.devices.items():
            try:
                func = getattr(device, method)
                results[device_id] = func(*args, **kwargs)
            except Exception as e:
                results[device_id] = {"error": str(e)}
        return results

    def collect_facts_all(self) -> Dict:
        """收集所有设备基本信息"""
        return self.execute_on_all("get_facts")

    def get_interfaces_all(self) -> Dict:
        """获取所有设备接口状态"""
        return self.execute_on_all("get_interfaces")

    def backup_all(
        self, output_dir: str = "backup/"
    ) -> Dict[str, str]:
        """备份所有设备配置"""
        import os
        from datetime import datetime

        os.makedirs(output_dir, exist_ok=True)
        results = {}

        for device_id, device in self.devices.items():
            try:
                config = device.get_config()
                timestamp = datetime.now().strftime(
                    "%Y%m%d_%H%M%S"
                )
                filename = os.path.join(
                    output_dir,
                    f"{device_id}_{timestamp}.cfg",
                )
                with open(filename, "w") as f:
                    f.write(config)
                results[device_id] = filename
            except Exception as e:
                results[device_id] = str(e)

        return results

四、使用示例

4.1 统一管理多厂商设备

#!/usr/bin/env python3
# demo_multi_vendor.py — 多厂商设备统一管理示例

from device_manager import UnifiedDeviceManager
import json


def main():
    # 创建统一管理器
    manager = UnifiedDeviceManager()

    # 添加各种厂商的设备
    devices_config = {
        "core-sw01": {
            "device_type": "huawei_vrp",
            "host": "192.168.1.1",
            "username": "admin",
            "password": "admin123",
        },
        "core-sw02": {
            "device_type": "cisco_ios",
            "host": "192.168.1.2",
            "username": "admin",
            "password": "admin123",
        },
        "acc-sw01": {
            "device_type": "huawei_vrp",
            "host": "192.168.1.10",
            "username": "admin",
            "password": "admin123",
        },
    }

    for dev_id, conn_info in devices_config.items():
        success = manager.add_device(dev_id, conn_info)
        print(f"{'✅' if success else '❌'} {dev_id}")

    # 统一收集设备信息（无需关心厂商差异）
    print("\n=== 设备基本信息 ===")
    facts = manager.collect_facts_all()
    for dev_id, fact in facts.items():
        if isinstance(fact, dict) and "error" in fact:
            print(f"{dev_id}: {fact['error']}")
        else:
            print(
                f"{dev_id}: {fact.hostname} "
                f"({fact.vendor} {fact.model})"
            )

    # 统一获取 BGP 信息
    print("\n=== BGP 信息 ===")
    bgp_all = manager.execute_on_all("get_bgp_peers")
    for dev_id, peers in bgp_all.items():
        if isinstance(peers, dict) and "error" in peers:
            print(f"{dev_id}: {peers['error']}")
        else:
            print(f"{dev_id}: {len(peers)} BGP peers")
            for peer in peers:
                print(f"  → {peer['peer_ip']} ({peer['state']})")

    # 统一备份配置
    print("\n=== 备份配置 ===")
    backups = manager.backup_all("backup/")
    for dev_id, path in backups.items():
        print(f"{dev_id}: {path}")

    # 断开所有连接
    for dev_id in list(manager.devices.keys()):
        manager.remove_device(dev_id)

    print("\n统一管理完成！")


if __name__ == "__main__":
    main()

4.2 跨厂商配置部署

# 跨厂商配置部署示例

def deploy_vlan_across_vendors(
    manager: UnifiedDeviceManager,
    vlan_id: int,
    vlan_name: str,
    interfaces: Dict[str, str],  # {device_id: interface_name}
):
    """跨厂商部署 VLAN"""

    vlan_commands = {
        "huawei_vrp": [
            f"vlan batch {vlan_id}",
            f"vlan {vlan_id}",
            f"name {vlan_name}",
        ],
        "cisco_ios": [
            f"vlan {vlan_id}",
            f"name {vlan_name}",
            "exit",
        ],
    }

    for dev_id, intf in interfaces.items():
        device = manager.get_device(dev_id)
        if not device:
            print(f"设备 {dev_id} 未连接")
            continue

        # 获取设备连接信息中的 device_type
        device_type = device.connection_info.get(
            "device_type", "huawei_vrp"
        )

        # 获取对应厂商的 VLAN 命令
        commands = vlan_commands.get(device_type, [])

        # 添加接口配置
        if device_type in ("huawei_vrp",):
            commands.extend([
                f"interface {intf}",
                f"port link-type access",
                f"port default vlan {vlan_id}",
            ])
        elif device_type in ("cisco_ios", "cisco_xe"):
            commands.extend([
                f"interface {intf}",
                f"switchport mode access",
                f"switchport access vlan {vlan_id}",
            ])

        # 执行配置
        print(f"在 {dev_id} ({device_type}) 上配置 VLAN {vlan_id}...")
        device.send_commands(commands)
        device.save_config()
        print(f"  ✅ {dev_id} 配置完成")

五、扩展新厂商适配器

5.1 适配器开发模板

#!/usr/bin/env python3
# new_vendor_adapter.py — 新厂商适配器开发模板

from device_abstraction import BaseDevice, DeviceFacts, \
    InterfaceInfo, RouteInfo
from typing import List, Dict, Optional


class NewVendorDevice(BaseDevice):
    """新厂商设备适配器（模板）"""

    def __init__(self, connection_info: Dict):
        super().__init__(connection_info)
        self._conn = None

    def connect(self) -> bool:
        """实现连接逻辑"""
        # TODO: 使用厂商特定的连接方式
        raise NotImplementedError

    def disconnect(self):
        """实现断开逻辑"""
        if self._conn:
            self._conn.close()

    def get_facts(self) -> DeviceFacts:
        """实现设备信息采集"""
        # TODO: 解析厂商特定的 show version 输出
        raise NotImplementedError

    def get_interfaces(self) -> Dict[str, InterfaceInfo]:
        """实现接口采集"""
        raise NotImplementedError

    def get_interface(
        self, name: str
    ) -> Optional[InterfaceInfo]:
        interfaces = self.get_interfaces()
        return interfaces.get(name)

    def get_bgp_peers(self) -> List[Dict]:
        """实现 BGP 采集"""
        raise NotImplementedError

    def get_routing_table(self) -> List[RouteInfo]:
        """实现路由表采集"""
        raise NotImplementedError

    def get_config(self) -> str:
        """实现配置采集"""
        raise NotImplementedError

    def send_commands(
        self, commands: List[str]
    ) -> List[str]:
        """实现命令下发"""
        raise NotImplementedError

    def save_config(self) -> bool:
        """实现配置保存"""
        raise NotImplementedError

    def get_lldp_neighbors(self) -> Dict:
        """实现 LLDP 采集"""
        raise NotImplementedError

    def ping(self, target: str, count: int = 3) -> Dict:
        """实现 Ping 测试"""
        raise NotImplementedError


# ── 注册新适配器 ──
def register():
    from device_manager import DeviceFactory
    DeviceFactory.register_adapter("new_vendor", NewVendorDevice)

六、最佳实践总结

多厂商统一管理最佳实践：

  1. 接口设计原则
  ┌─ 接口方法粒度适中（不要太细也不要太粗）
  ├─ 返回统一的数据类型（不是原始文本）
  ├─ 异常处理标准化（统一异常类）
  └─ 支持超时和重试机制

  2. 适配器开发
  ┌─ 每个厂商一个适配器类
  ├─ 适配器只负责协议转换（不包含业务逻辑）
  ├─ 输出解析尽可能健壮（应对版本差异）
  └─ 适配器单元测试覆盖主要命令

  3. 厂商差异处理
  ┌─ 命令转换：通过适配器封装
  ├─ 输出解析：正则表达式 + 结构化映射
  ├─ 能力声明：声明设备支持/不支持的能力
  └─ 降级策略：不支持的能力返回友好提示

  4. 扩展性
  ┌─ 新厂商只需实现适配器接口
  ├─ 注册到工厂即可使用
  ├─ 不修改现有代码（开闭原则）
  └─ 适配器可独立发布和版本管理

  5. 生产使用
  ┌─ 连接池管理（复用连接）
  ├─ 并发控制（不超限）
  ├─ 详细日志（便于排查问题）
  └─ 监控适配器自身状态

下篇预告：第327篇 — 网络自动化运维平台架构设计，从架构角度设计企业级网络自动化运维平台。

下篇预告：第327篇 — 网络自动化运维平台架构设计，从架构角度设计企业级网络自动化运维平台。