第306篇：Jinja2 模板引擎——配置模板化

关键词：Jinja2、模板引擎、配置模板、宏、继承、条件渲染、循环、网络配置生成

一、引言

在前几篇中，我们掌握了连接设备、批量执行命令的方法。但当配置内容变得复杂——比如为50台交换机分别生成包含VLAN、接口、BGP、OSPF等数十行配置的完整配置时，逐条拼接字符串的方式既难以维护又容易出错。

Jinja2是Python生态中最强大的模板引擎，Flask、Ansible、OpenStack等知名项目均以其为核心。它将配置内容从"硬编码命令"提升为"参数化模板"，使一份模板即可适配数百台设备的差异化配置需求。

二、Jinja2 基础语法

2.1 安装与Hello World

pip install jinja2

from jinja2 import Template

# 最基本的模板渲染
template_str = "你好, {{ name }}！欢迎学习 {{ topic }}。"
template = Template(template_str)

output = template.render(name="网络工程师", topic="Jinja2")
print(output)
# 输出: 你好, 网络工程师！欢迎学习 Jinja2。

2.2 核心语法速查

Jinja2使用三种分隔符：

三、条件渲染与循环

3.1 条件判断（if/elif/else）

from jinja2 import Template

template_str = """
hostname {{ hostname }}

{% if role == "core" %}
! === 核心交换机配置 ===
vlan batch 10 20 30 100 200
ospf 1 router-id {{ router_id }}
 area 0.0.0.0
  network 10.0.0.0 0.0.0.255
{% elif role == "aggregation" %}
! === 汇聚交换机配置 ===
vlan batch 10 20 30
ospf 1 router-id {{ router_id }}
 area 0.0.0.0
  network {{ mgmt_network }} 0.0.0.255
{% elif role == "access" %}
! === 接入交换机配置 ===
vlan batch 10 20
{% endif %}
"""

template = Template(template_str)

# 渲染核心交换机配置
core_config = template.render(
    hostname="Core-SW-01",
    role="core",
    router_id="10.255.0.1",
    mgmt_network="10.255.1.0"
)
print(core_config)

# 渲染接入交换机配置
access_config = template.render(
    hostname="Access-SW-01",
    role="access",
    router_id="10.255.2.1"
)
print(access_config)

3.2 for 循环

from jinja2 import Template

template_str = """
! === VLAN 配置 ===
{% for vlan_id, vlan_info in vlans.items() %}
vlan {{ vlan_id }}
 description {{ vlan_info.name }}
{% if vlan_info.ipv4_subnet %}
 ip interface {{ vlan_info.ipv4_subnet }}
{% endif %}
{% endfor %}

! === 接口配置 ===
{% for iface in interfaces %}
interface {{ iface.name }}
 description {{ iface.description }}
 port link-type {{ iface.link_type }}
{% if iface.pvid %}
 port default vlan {{ iface.pvid }}
{% endif %}
{% if iface.trunk_vlans %}
 port trunk allow-pass vlan {{ iface.trunk_vlans }}
{% endif %}
{% endfor %}
"""

template = Template(template_str)

# 渲染数据
config = template.render(
    vlans={
        10: {"name": "USER_VLAN", "ipv4_subnet": "10.10.10.1/24"},
        20: {"name": "SERVER_VLAN", "ipv4_subnet": "10.20.20.1/24"},
        100: {"name": "MGMT_VLAN", "ipv4_subnet": None},
    },
    interfaces=[
        {"name": "GigabitEthernet0/0/1", "description": "To-Core-01",
         "link_type": "trunk", "pvid": None, "trunk_vlans": "10 20 100"},
        {"name": "GigabitEthernet0/0/2", "description": "To-Core-02",
         "link_type": "trunk", "pvid": None, "trunk_vlans": "10 20 100"},
        {"name": "GigabitEthernet0/0/10", "description": "To-Server-A",
         "link_type": "access", "pvid": 20, "trunk_vlans": None},
    ]
)
print(config)

3.3 循环中的过滤器与特殊变量

template_str = """
{% for iface in interfaces %}
! 接口 {{ loop.index }}/{{ loop.length }}: {{ iface.name }}
{% if loop.first %}
! ---- 第一个接口（通常是上行） ----
{% endif %}
{% if loop.last %}
! ---- 最后一个接口 ----
{% endif %}
{% endfor %}
"""

# loop.index     - 当前索引（从1开始）
# loop.index0    - 当前索引（从0开始）
# loop.first     - 是否为第一个元素
# loop.last      - 是否为最后一个元素
# loop.length    - 总元素数

四、宏（Macro）：配置块复用

宏类似于Python中的函数，定义一次、多次调用：

from jinja2 import Template

template_str = """
{# 定义宏：生成接口配置块 #}
{% macro config_interface(iface) %}
interface {{ iface.name }}
 description {{ iface.description }}
{% if iface.link_type == "access" %}
 port link-type access
 port default vlan {{ iface.pvid }}
{% elif iface.link_type == "trunk" %}
 port link-type trunk
 port trunk allow-pass vlan {{ iface.trunk_vlans }}
{% endif %}
{% if iface.ip_address %}
 ip address {{ iface.ip_address }}
{% endif %}
{% endmacro %}

{# 定义宏：生成VLAN配置块 #}
{% macro config_vlan(vlan_id, name, ipv4=None) %}
vlan {{ vlan_id }}
 description {{ name }}
{% if ipv4 %}
 ip interface {{ ipv4 }}
{% endif %}
{% endmacro %}

! ========== VLAN配置 ==========
{{ config_vlan(10, "USER", "10.10.10.1/24") }}
{{ config_vlan(20, "SERVER", "10.20.20.1/24") }}
{{ config_vlan(100, "MGMT") }}

! ========== 接口配置 ==========
{{ config_interface(interfaces[0]) }}
{{ config_interface(interfaces[1]) }}
{{ config_interface(interfaces[2]) }}
"""

template = Template(template_str)
config = template.render(interfaces=[
    {"name": "GE0/0/1", "description": "Uplink-Core",
     "link_type": "trunk", "trunk_vlans": "10-20 100", "ip_address": None},
    {"name": "GE0/0/10", "description": "To-Server-Rack",
     "link_type": "access", "pvid": 20, "ip_address": None},
    {"name": "Vlanif100", "description": "Management",
     "link_type": "access", "pvid": 100, "ip_address": "10.255.0.1/24"},
])
print(config)

五、模板继承：避免重复配置

模板继承允许定义一个基础模板，子类模板只需覆盖特定区块：

from jinja2 import Environment, FileSystemLoader
import os

# 使用Environment管理多个模板
env = Environment(loader=FileSystemLoader("templates"))

# ============================================================
# 基础模板: base_config.j2
# ============================================================
base_template = """
! ============================================================
! 设备配置 - {{ hostname }}
! 生成时间: {{ generated_at }}
! ============================================================

hostname {{ hostname }}

{% block sys_opt %}
sysname {{ hostname }}
clock timezone Beijing add 8
{% endblock %}

! --- SSH 配置 ---
{% block ssh_config %}
ssh server enable
aaa
 authentication-scheme default
  scheme password
 authorization-scheme default
  scheme authenticated
 domain default
  authentication-scheme default
  authorization-scheme default
user-interface vty 0 4
 authentication-mode aaa
 protocol inbound ssh
{% endblock %}

! --- 业务配置 ---
{% block business_config %}
{# 子类覆盖此区块 #}
{% endblock %}

! --- 保存配置 ---
return
save force
"""

# ============================================================
# 接入交换机模板: access_switch.j2 （继承base_config）
# ============================================================
access_template = """
{% extends "base_config.j2" %}

{% block business_config %}
! === 接入层VLAN ===
{% for vlan_id, vlan_name in vlans.items() %}
vlan {{ vlan_id }}
 description {{ vlan_name }}
{% endfor %}

! === 接入层接口 ===
{% for iface in interfaces %}
interface {{ iface.name }}
 description {{ iface.description }}
 port link-type {{ iface.link_type }}
{% if iface.link_type == "access" %}
 port default vlan {{ iface.pvid }}
{% elif iface.link_type == "trunk" %}
 port trunk allow-pass vlan {{ iface.trunk_vlans }}
{% endif %}
{% endfor %}

! === STP 配置 ===
stp mode mstp
stp region-configuration
 region-name {{ stp_region }}
{% for vlan_id in stp_instances %}
 instance {{ vlan_id }}
{% endfor %}
 active region-configuration
stp edged-port default
{% endblock %}
"""

# ============================================================
# 渲染接入交换机配置
# ============================================================
from datetime import datetime

# 实际使用时将模板保存到templates/目录，此处用字符串演示
access_config = access_template
# 简化演示，实际应使用FileSystemLoader

六、BGP配置模板实战

6.1 完整BGP模板

from jinja2 import Template

bgp_template = Template("""
! ===== BGP 配置 =====
bgp {{ local_as }}
 router-id {{ router_id }}
 !
 {% if graceful_restart %}
 graceful-restart
 {% endif %}
 !
 {% for afi_safi, config in address_families.items() %}
 address-family {{ afi_safi }}
 {% endfor %}
 !
{% for peer_group in peer_groups %}
{% if peer_group.name %}
 peer {{ peer_group.name }} group
  peer as-number {{ peer_group.remote_as }}
{% if peer_group.password %}
  peer password simple {{ peer_group.password }}
{% endif %}
{% if peer_group.description %}
  peer description {{ peer_group.description }}
{% endif %}
{% if peer_group.send_community %}
  peer enable-ipv6 send-community
{% endif %}
{% for afi_safi in peer_group.address_families %}
  peer {{ peer_group.name }} allow-import-route address-family {{ afi_safi }}
  peer {{ peer_group.name }} enable address-family {{ afi_safi }}
{% endfor %}
{% endif %}
{% for peer in peer_group.get('peers', []) %}
 peer {{ peer.ip }}
  peer as-number {{ peer.asn | default(peer_group.remote_as) }}
{% if peer.description %}
  peer description {{ peer.description }}
{% endif %}
{% if peer.local_ip %}
  peer bind-peers {{ peer.local_ip }}
{% endif %}
{% if peer.next_hop_self %}
  peer enable-ipv6 next-update-origin
{% endif %}
{% for afi_safi in peer_group.address_families %}
  peer {{ peer.ip }} allow-import-route address-family {{ afi_safi }}
  peer {{ peer.ip }} enable address-family {{ afi_safi }}
{% endfor %}
{% endfor %}
{% endfor %}
 !
{% for import_policy in import_policies %}
 import-route policy {{ import_policy.name }} address-family {{ import_policy.afi_safi }}
{% endfor %}
{% for export_policy in export_policies %}
 export-route policy {{ export_policy.name }} address-family {{ export_policy.afi_safi }}
{% endfor %}
""")

# 渲染BGP配置
bgp_config = bgp_template.render(
    local_as=65001,
    router_id="10.255.0.1",
    graceful_restart=True,
    address_families={
        "ipv4-unicast": {},
        "ipv6-unicast": {},
        "labeled-ipv4-unicast": {},
    },
    peer_groups=[
        {
            "name": "INTERNAL_PEERS",
            "remote_as": 65001,
            "description": "iBGP Peers",
            "send_community": True,
            "address_families": ["ipv4-unicast", "labeled-ipv4-unicast"],
            "peers": [
                {"ip": "10.255.0.2", "description": "To-PE-02"},
                {"ip": "10.255.0.3", "description": "To-PE-03"},
            ]
        },
        {
            "name": "EBGP_PEERS",
            "remote_as": 65100,
            "password": "BgpP@ss2025",
            "description": "eBGP Peers",
            "address_families": ["ipv4-unicast"],
            "peers": [
                {"ip": "203.0.113.1", "description": "To-ISP-A"},
            ]
        }
    ],
    import_policies=[
        {"name": "IMPORT_DEFAULT", "afi_safi": "ipv4-unicast"}
    ],
    export_policies=[
        {"name": "EXPORT_TO_EBGP", "afi_safi": "ipv4-unicast"},
    ]
)
print(bgp_config)

七、模板变量管理与多设备适配

7.1 从YAML读取设备变量

import yaml
from jinja2 import Environment, FileSystemLoader

# 设备变量文件: devices.yaml
devices_yaml = """
devices:
  - hostname: Core-SW-01
    role: core
    mgmt_ip: 10.255.0.1
    router_id: 10.255.0.1
    vlans:
      10: {name: USER, subnet: 10.10.10.1/24}
      20: {name: SERVER, subnet: 10.20.20.1/24}
      100: {name: MGMT, subnet: null}
    uplinks:
      - name: GigabitEthernet0/0/1
        description: To-Core-SW-02
        type: trunk
        vlans: "10-20 100"
      - name: GigabitEthernet0/0/2
        description: To-Core-SW-03
        type: trunk
        vlans: "10-20 100"
    local_vlans:
      - {name: GigabitEthernet0/0/10, pvid: 10, desc: To-Server-A}
      - {name: GigabitEthernet0/0/11, pvid: 20, desc: To-Server-B}
  - hostname: Core-SW-02
    role: core
    mgmt_ip: 10.255.0.2
    router_id: 10.255.0.2
    vlans:
      10: {name: USER, subnet: 10.10.10.2/24}
      20: {name: SERVER, subnet: 10.20.20.2/24}
      100: {name: MGMT, subnet: null}
    uplinks:
      - name: GigabitEthernet0/0/1
        description: To-Core-SW-01
        type: trunk
        vlans: "10-20 100"
    local_vlans:
      - {name: GigabitEthernet0/0/10, pvid: 10, desc: To-Server-C}
"""

# 解析YAML
data = yaml.safe_load(devices_yaml)

# 加载模板
env = Environment(loader=FileSystemLoader("templates"))
template = env.get_template("switch_config.j2")

# 为每台设备生成配置
for device in data["devices"]:
    config = template.render(**device)
    filename = f"configs/{device['hostname']}.cfg"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(config)
    print(f"[+] 已生成 {filename}")

7.2 自定义过滤器

from jinja2 import Environment

env = Environment()

# 自定义过滤器：将VLAN ID列表转为范围字符串
def vlan_range(vlan_list):
    """将 [10, 11, 12, 20, 21, 100] 转为 '10-12 20-21 100'"""
    if not vlan_list:
        return ""
    vlan_list = sorted(set(vlan_list))
    ranges = []
    start = vlan_list[0]
    end = vlan_list[0]

    for vlan in vlan_list[1:]:
        if vlan == end + 1:
            end = vlan
        else:
            if start == end:
                ranges.append(str(start))
            else:
                ranges.append(f"{start}-{end}")
            start = vlan
            end = vlan

    if start == end:
        ranges.append(str(start))
    else:
        ranges.append(f"{start}-{end}")

    return " ".join(ranges)

# 注册过滤器
env.filters["vlan_range"] = vlan_range

# 使用
template = env.from_string("port trunk allow-pass vlan {{ vlans | vlan_range }}")
output = template.render(vlans=[10, 11, 12, 20, 21, 100])
print(output)  # port trunk allow-pass vlan 10-12 20-21 100

八、故障案例：模板变量未定义导致静默错误

8.1 故障现象

批量生成的50台设备配置中，10台缺失BGP配置段。

8.2 排查过程

模板片段：

{% if bgp_enabled %}
bgp {{ bgp_local_as }}
 router-id {{ bgp_router_id }}
 peer {{ bgp_peer_ip }}
  peer as-number {{ bgp_peer_as }}
{% endif %}

设备变量中bgp_enabled字段缺失时，Jinja2默认将其视为undefined，在某些环境下undefined等价于False，导致{% if %}块被静默跳过，不报错也不告警。

8.3 解决方案

from jinja2 import Environment, UndefinedError, StrictUndefined

# 方案1：严格模式——未定义变量直接报错
env = Environment(
    loader=FileSystemLoader("templates"),
    undefined=StrictUndefined   # 未定义变量抛出异常
)

# 方案2：默认值
template = Template("""
{% if bgp_enabled is defined and bgp_enabled %}
bgp {{ bgp_local_as }}
{% endif %}
""")

# 方案3：使用default过滤器
template = Template("""
{% if bgp_enabled | default(false) %}
bgp {{ bgp_local_as | default(65000) }}
{% endif %}
""")

八（续）、IOS设备自动化与批量验证

8.1 IOS 设备自动化脚本

#!/usr/bin/env python3
# scripts/ios_automation.py
"""
Cisco IOS 设备自动化脚本
支持：IOS/IOS-XE 设备配置生成、批量下发、验证
"""
from jinja2 import Environment, FileSystemLoader, StrictUndefined
from netmiko import ConnectHandler
import yaml
import logging
from pathlib import Path
from typing import Dict, List

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)


class IOSConfigGenerator:
    """IOS 配置生成器"""

    def __init__(self, template_dir: str = 'templates/ios'):
        self.env = Environment(
            loader=FileSystemLoader(template_dir),
            undefined=StrictUndefined
        )
        # 注册自定义过滤器
        self.env.filters['ios_vlan_range'] = self.ios_vlan_range
        self.env.filters['ios_acl_action'] = self.ios_acl_action
        self.env.filters['ios_wildcard'] = self.ios_wildcard_mask

    def ios_vlan_range(self, vlan_ids: List[int]) -> str:
        """VLAN ID 列表转为 Cisco 范围格式: 10,11,12,20-25"""
        if not vlan_ids:
            return ''
        sorted_ids = sorted(set(vlan_ids))
        ranges = []
        start = end = sorted_ids[0]
        for vid in sorted_ids[1:]:
            if vid == end + 1:
                end = vid
            else:
                ranges.append(str(start) if start == end else f'{start}-{end}')
                start = end = vid
        ranges.append(str(start) if start == end else f'{start}-{end}')
        return ','.join(ranges)

    def ios_acl_action(self, action: str) -> str:
        """转换 ACL 动作"""
        return 'permit' if action.lower() == 'permit' else 'deny'

    def ios_wildcard_mask(self, cidr: str) -> str:
        """CIDR 掩码转通配符掩码"""
        if '/' in cidr:
            prefix_len = int(cidr.split('/')[1])
        else:
            prefix_len = 24
        mask = (0xFFFFFFFF >> prefix_len) & 0xFFFFFFFF
        return '.'.join(str((mask >> (8 * i)) & 0xFF) for i in range(3, -1, -1))

    def generate(self, template_name: str, **kwargs) -> str:
        """生成配置"""
        template = self.env.get_template(template_name)
        return template.render(**kwargs)


# IOS 模板示例
IOS_BASE_TEMPLATE = """
! ============================================
! 设备: {{ hostname }}
! 角色: {{ role }}
! 生成时间: {{ generated_at }}
! ============================================

hostname {{ hostname }}

! --- 系统优化 ---
ip domain-name {{ domain_name | default('localdomain') }}
ip name-server {{ dns_primary }} {{ dns_secondary | default('8.8.8.8') }}
no ip domain-lookup

! --- AAA 配置 ---
aaa new-model
aaa authentication login default local
aaa authorization exec default local
aaa session-id common

! --- 用户管理 ---
username {{ admin_user }} privilege 15 secret {{ admin_password }}

! --- SSH 配置 ---
ip ssh version 2
ip ssh maxstartups 10
ip ssh time-out 60
ip ssh authentication-retries 3
line vty 0 15
 login authentication default
 transport input ssh
 no exec-timo

! --- NTP 配置 ---
{% for server in ntp_servers %}
ntp server {{ server }}
{% endfor %}

! --- SNMP 配置 ---
snmp-server community {{ snmp_ro_community }} RO
snmp-server host {{ snmp_trap_host }} traps version 2c {{ snmp_community }}

! --- 日志配置 ---
logging host {{ syslog_host }}
logging trap informational

! --- 业务配置 ---
{% block business_config %}{% endblock %}

! --- 保存 ---
end
write memory
"""

IOS_INTERFACE_TEMPLATE = """
{% for iface in interfaces %}
! === {{ iface.name }}: {{ iface.description }} ===
interface {{ iface.name }}
 description {{ iface.description }}
{% if iface.ip_address %}
 ip address {{ iface.ip_address }}
{% endif %}
{% if iface.secondary_ips %}
{% for sec_ip in iface.secondary_ips %}
 ip address {{ sec_ip }} secondary
{% endfor %}
{% endif %}
{% if iface.duplicate %}
 ip address {% for ip in iface.duplicate %}{{ ip }}{% if not loop.last %} {{ ' ' }}{% endif %}{% endfor %}
{% endif %}
{% if iface.native_vlan %}
 switchport mode access
 switchport access vlan {{ iface.native_vlan }}
{% elif iface.trunk_vlans %}
 switchport mode trunk
 switchport trunk allowed vlan {{ iface.trunk_vlans | ios_vlan_range }}
{% endif %}
{% if iface.speed %}
 speed {{ iface.speed }}
{% endif %}
{% if iface.duplex %}
 duplex {{ iface.duplex }}
{% endif %}
{% if iface.mtu %}
 mtu {{ iface.mtu }}
{% endif %}
{% if iface.shutdown is sameas false %}
 no shutdown
{% endif %}
!
{% endfor %}
"""

IOS_BGP_TEMPLATE = """
! === BGP 配置 ===
router bgp {{ local_as }}
 bgp router-id {{ router_id }}
 bgp log-neighbors-changes
{% if graceful_restart %}
 bgp graceful-restart
 bgp graceful-restart restart-time {{ gr_restart_time | default(120) }}
 bgp graceful-restart helper
{% endif %}
!
{% for peer_group in peer_groups %}
! --- Peer Group: {{ peer_group.name }} ---
 neighbor {{ peer_group.name }} peer group
 neighbor {{ peer_group.name }} remote-as {{ peer_group.remote_as }}
{% if peer_group.password %}
 neighbor {{ peer_group.name }} password {{ peer_group.password }}
{% endif %}
{% if peer_group.description %}
 neighbor {{ peer_group.name }} description {{ peer_group.description }}
{% endif %}
 neighbor {{ peer_group.name }} timers {{ peer_group.timers_keepalive | default(10) }} {{ peer_group.timers_holdtime | default(30) }}
{% if peer_group.update_source %}
 neighbor {{ peer_group.name }} update-source {{ peer_group.update_source }}
{% endif %}
{% for af in peer_group.address_families %}
 address-family {{ af }}
  neighbor {{ peer_group.name }} activate
{% if peer_group.route_maps %}
  neighbor {{ peer_group.name }} route-map {{ peer_group.route_maps.get('import', 'IMPORT_' ~ peer_group.name) }} in
  neighbor {{ peer_group.name }} route-map {{ peer_group.route_maps.get('export', 'EXPORT_' ~ peer_group.name) }} out
{% endif %}
{% if peer_group.send_community %}
  neighbor {{ peer_group.name }} send-community
{% endif %}
 exit-address-family
{% endfor %}
!
{% for peer in peer_group.get('members', []) %}
 neighbor {{ peer.ip }} peer group {{ peer_group.name }}
{% if peer.description %}
 neighbor {{ peer.ip }} description {{ peer.description }}
{% endif %}
{% if peer.local_ip %}
 neighbor {{ peer.ip }} update-source {{ peer.local_ip }}
{% endif %}
!
{% endfor %}
{% endfor %}
!
{% for network in networks %}
 network {{ network }}
{% endfor %}
!
{% for redistribute in redistribute_routes %}
 redistribute {{ redistribute.protocol }} metric {{ redistribute.metric | default(100) }}
{% endfor %}
"""

IOS_ACL_TEMPLATE = """
! === ACL 配置 ===
{% for acl in acls %}
! --- ACL: {{ acl.name }} ({{ acl.type }}) ---
ip access-list {{ acl.type }} {{ acl.name }}
{% for rule in acl.rules %}
 {{ rule.action | lower }} {{ rule.protocol | default('ip') }} {{ rule.source }} {{ rule.source_mask }} {{ rule.destination }} {{ rule.dest_mask }}{% if rule.port %} eq {{ rule.port }}{% endif %}
{% endfor %}
!
{% endfor %}
"""

# 使用示例
if __name__ == '__main__':
    gen = IOSConfigGenerator()

    # 加载设备参数
    with open('devices/ios_pe_router.yaml') as f:
        params = yaml.safe_load(f)

    # 生成配置
    config = gen.generate('ios_full_config.j2', **params)
    print(config)

8.2 配置模板引擎（统一多厂商）

#!/usr/bin/env python3
# scripts/multi_vendor_engine.py
"""
多厂商配置模板引擎
根据设备平台自动选择对应模板
"""
from jinja2 import Environment, FileSystemLoader, StrictUndefined, ChoiceLoader, PrefixLoader
import yaml
import logging
from pathlib import Path
from typing import Dict, List

logger = logging.getLogger(__name__)

# 平台到模板前缀的映射
PLATFORM_TEMPLATE_MAP = {
    'huawei_vrpv8': 'huawei',
    'cisco_ios': 'cisco_ios',
    'cisco_xr': 'cisco_xr',
    'juniper_junos': 'juniper',
    'arista_eos': 'arista',
}

# 统一参数模型
UNIFIED_CONFIG_MODEL = {
    'hostname': str,
    'interfaces': List[Dict],
    'vlans': Dict,
    'bgp': Dict,
    'ospf': Dict,
    'isis': Dict,
    'acls': List[Dict],
    'routing_policies': List[Dict],
    'snmp': Dict,
    'ntp': Dict,
    'syslog': Dict,
}


class MultiVendorConfigEngine:
    """多厂商配置模板引擎"""

    def __init__(self, template_root: str = 'templates'):
        self.template_root = Path(template_root)

        # 构建前缀加载器
        loaders = {}
        for prefix in PLATFORM_TEMPLATE_MAP.values():
            prefix_dir = self.template_root / prefix
            if prefix_dir.exists():
                loaders[prefix] = FileSystemLoader(str(prefix_dir))

        # 公共模板（跨厂商共用）
        loaders['common'] = FileSystemLoader(str(self.template_root / 'common'))

        self.env = Environment(
            loader=ChoiceLoader(list(loaders.values())),
            undefined=StrictUndefined
        )

        # 注册通用过滤器
        self._register_filters()

    def _register_filters(self):
        """注册通用过滤器"""
        self.env.filters['vlan_range'] = self._vlan_range_filter
        self.env.filters['sort_by_name'] = self._sort_by_name_filter
        self.env.filters['first_ip'] = self._first_ip_filter

    @staticmethod
    def _vlan_range_filter(vlans: List[int]) -> str:
        """VLAN 范围格式"""
        if not vlans:
            return ''
        sorted_v = sorted(set(vlans))
        ranges = []
        s = e = sorted_v[0]
        for v in sorted_v[1:]:
            if v == e + 1:
                e = v
            else:
                ranges.append(str(s) if s == e else f'{s}-{e}')
                s = e = v
        ranges.append(str(s) if s == e else f'{s}-{e}')
        return ' '.join(ranges)

    def get_template_prefix(self, platform: str) -> str:
        """获取平台对应的模板前缀"""
        return PLATFORM_TEMPLATE_MAP.get(platform, 'huawei')

    def generate_config(self, device: Dict,
                       template_name: str = 'full_config') -> str:
        """
        为单台设备生成配置
        :param device: 设备参数（统一模型）
        :param template_name: 模板名称（不含前缀）
        """
        prefix = self.get_template_prefix(device.get('platform', 'huawei_vrpv8'))
        tpl_name = f"{prefix}/{template_name}.j2"

        try:
            template = self.env.get_template(tpl_name)
            config = template.render(**device)
            logger.info(f"为设备 {device.get('hostname')} 生成配置 "
                       f"(平台: {prefix}, 模板: {template_name})")
            return config
        except Exception as e:
            logger.error(f"模板渲染失败: {e}")
            raise

    def generate_all(self, devices_file: str,
                    output_dir: str = 'generated_configs') -> List[str]:
        """批量生成所有设备配置"""
        with open(devices_file) as f:
            data = yaml.safe_load(f)

        devices = data.get('devices', [])
        output_path = Path(output_dir)
        output_path.mkdir(parents=True, exist_ok=True)

        generated = []
        for device in devices:
            try:
                config = self.generate_config(device)
                filename = output_path / f"{device['hostname']}.cfg"
                filename.write_text(config, encoding='utf-8')
                generated.append(str(filename))
            except Exception as e:
                logger.error(f"设备 {device.get('hostname')} 配置生成失败: {e}")

        logger.info(f"批量生成完成: {len(generated)}/{len(devices)} 台成功")
        return generated

8.3 IOS 批量验证脚本

#!/usr/bin/env python3
# scripts/ios_validator.py
"""
IOS 配置批量验证脚本
在配置下发前进行语法和逻辑校验
"""
import re
import logging
from typing import Dict, List, Tuple
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class ValidationResult:
    """验证结果"""
    config_name: str
    passed: bool
    errors: List[str] = None
    warnings: List[str] = None

    def __post_init__(self):
        self.errors = self.errors or []
        self.warnings = self.warnings or []


class IOSConfigValidator:
    """IOS 配置验证器"""

    # 验证规则
    RULES = {
        'hostname_format': (
            r'^hostname [a-zA-Z0-9][a-zA-Z0-9\-\.]*$',
            '主机名格式不符合规范'
        ),
        'no_telnet': (
            r'line vty.*transport input (telnet|all)',
            '不建议使用 Telnet，应仅使用 SSH'
        ),
        'snmp_community': (
            r'snmp-server community\s+\S+\s+?(?:RO|RW)?$',
            'SNMP Community 应使用 SNMPv3'
        ),
        'no_cdp': (
            r'cdp\s+run',
            '生产环境建议关闭 CDP'
        ),
        'bgp_as_format': (
            r'router bgp \d+',
            None  # 仅做格式检查，不报错
        ),
    }

    def __init__(self):
        self.compiled_rules = {}
        for name, (pattern, message) in self.RULES.items():
            self.compiled_rules[name] = (re.compile(pattern), message)

    def validate(self, config: str, config_name: str = '') -> ValidationResult:
        """验证配置"""
        result = ValidationResult(
            config_name=config_name or 'unknown',
            passed=True
        )

        lines = config.splitlines()

        for name, (pattern, message) in self.compiled_rules.items():
            for line_num, line in enumerate(lines, 1):
                if pattern.search(line):
                    if message:
                        result.warnings.append(
                            f"第 {line_num} 行: {message} "
                            f"(匹配: {line.strip()[:50]})"
                        )

        # 检查必要的配置段
        required_sections = {
            'hostname': '缺少 hostname 配置',
            'aaa new-model': '缺少 AAA 配置',
            'ip ssh version': '缺少 SSH 版本配置',
        }

        config_text = '\n'.join(lines)
        for section, error_msg in required_sections.items():
            if section not in config_text:
                result.errors.append(f"[缺少] {error_msg}")
                result.passed = False

        # 检查 IP 地址格式
        ip_pattern = re.compile(r'ip address\s+(\S+)\s+(\S+)')
        for line_num, line in enumerate(lines, 1):
            match = ip_pattern.search(line)
            if match:
                ip = match.group(1)
                mask = match.group(2)
                if not self._valid_ip(ip):
                    result.errors.append(
                        f"第 {line_num} 行: 无效 IP 地址 '{ip}'"
                    )
                    result.passed = False
                if not self._valid_mask(mask):
                    result.errors.append(
                        f"第 {line_num} 行: 无效掩码 '{mask}'"
                    )
                    result.passed = False

        # 检查 ACL 规则完整性
        acl_errors = self._validate_acls(lines)
        result.errors.extend(acl_errors)
        if acl_errors:
            result.passed = False

        return result

    @staticmethod
    def _valid_ip(ip: str) -> bool:
        """验证 IP 地址格式"""
        parts = ip.split('.')
        if len(parts) != 4:
            return False
        try:
            return all(0 <= int(p) <= 255 for p in parts)
        except ValueError:
            return False

    @staticmethod
    def _valid_mask(mask: str) -> bool:
        """验证子网掩码格式"""
        if '/' in mask:
            try:
                return 0 <= int(mask.split('/')[1]) <= 32
            except ValueError:
                return False
        parts = mask.split('.')
        if len(parts) != 4:
            return False
        try:
            return all(0 <= int(p) <= 255 for p in parts)
        except ValueError:
            return False

    def _validate_acls(self, lines: List[str]) -> List[str]:
        """验证 ACL 规则"""
        errors = []
        in_acl = False
        acl_name = ''

        for line_num, line in enumerate(lines, 1):
            stripped = line.strip()
            if stripped.startswith('ip access-list'):
                in_acl = True
                parts = stripped.split()
                if len(parts) >= 4:
                    acl_name = parts[-1]
                continue

            if in_acl and stripped.startswith('!'):
                in_acl = False
                continue

        return errors

    def validate_batch(self, configs: Dict[str, str]) -> List[ValidationResult]:
        """批量验证配置"""
        results = []
        for name, config in configs.items():
            result = self.validate(config, name)
            results.append(result)
            status = '✅ 通过' if result.passed else '❌ 失败'
            logger.info(f"验证 {name}: {status} "
                       f"(错误: {len(result.errors)}, "
                       f"警告: {len(result.warnings)})")
        return results

8.4 IOS 特定命令处理

#!/usr/bin/env python3
# scripts/ios_command_handler.py
"""
IOS 特定命令处理
处理 Cisco IOS/IOS-XE 特有的命令语法和行为
"""
import re
from typing import Dict, List, Tuple, Optional


class IOSCommandHandler:
    """IOS 命令处理器"""

    # IOS 特有命令映射
    IOS_COMMANDS = {
        # 接口相关
        'enable_port': 'no shutdown',
        'disable_port': 'shutdown',
        'set_description': 'description {desc}',
        'set_ip': 'ip address {ip} {mask}',
        'set_vlan_access': 'switchport access vlan {vlan}',
        'set_vlan_trunk': 'switchport trunk allowed vlan {vlans}',
        'set_native_vlan': 'switchport access vlan {vlan}',
        'set_speed': 'speed {speed}',
        'set_duplex': 'duplex {duplex}',
        'set_mtu': 'mtu {mtu}',

        # VLAN 相关
        'create_vlan': 'vlan {vlan_id}\n name {vlan_name}',
        'delete_vlan': 'no vlan {vlan_id}',

        # 路由相关
        'static_route': 'ip route {dest} {mask} {next_hop}',
        'ospf_network': 'network {network} {wildcard} area {area}',

        # ACL 相关
        'permit_acl': 'permit {protocol} {src} {src_mask} {dst} {dst_mask}',
        'deny_acl': 'deny {protocol} {src} {src_mask} {dst} {dst_mask}',
    }

    # IOS 命令输出解析模式
    OUTPUT_PATTERNS = {
        'interface_status': re.compile(
            r'(\S+)\s+[\w\-]+\s+(\S+)\s+(\S+)\s+(\S+)'
        ),
        'bgp_peer': re.compile(
            r'\s+(\d+\.\d+\.\d+\.\d+)\s+\S+\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)'
        ),
        'ospf_neighbor': re.compile(
            r'\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)'
        ),
        'arp_entry': re.compile(
            r'Internet\s+(\d+\.\d+\.\d+\.\d+)\s+\d+\s+([\da-f:]+)\s+\S+\s+(\S+)'
        ),
    }

    def translate_command(self, command: str, **kwargs) -> str:
        """
        将通用命令翻译为 IOS 命令
        :param command: 命令模板名
        :param kwargs: 命令参数
        """
        template = self.IOS_COMMANDS.get(command)
        if not template:
            raise ValueError(f"未知命令: {command}")
        return template.format(**kwargs)

    def parse_show_interfaces(self, output: str) -> List[Dict]:
        """解析 show interfaces 输出"""
        interfaces = []
        for line in output.splitlines():
            match = self.OUTPUT_PATTERNS['interface_status'].search(line)
            if match:
                interfaces.append({
                    'name': match.group(1),
                    'protocol': match.group(2),
                    'status': match.group(3),
                    'duplex': match.group(4),
                })
        return interfaces

    def parse_show_bgp(self, output: str) -> List[Dict]:
        """解析 show ip bgp summary 输出"""
        peers = []
        for line in output.splitlines():
            match = self.OUTPUT_PATTERNS['bgp_peer'].search(line)
            if match:
                peers.append({
                    'ip': match.group(1),
                    'as': match.group(2),
                    'msg_rcvd': match.group(3),
                    'msg_sent': match.group(4),
                    'state': match.group(5),
                })
        return peers

    def parse_show_arp(self, output: str) -> List[Dict]:
        """解析 show ip arp 输出"""
        entries = []
        for line in output.splitlines():
            match = self.OUTPUT_PATTERNS['arp_entry'].search(line)
            if match:
                entries.append({
                    'ip': match.group(1),
                    'mac': match.group(2),
                    'interface': match.group(3),
                })
        return entries

    def generate_ios_commands(self, actions: List[Dict]) -> List[str]:
        """
        根据操作列表生成 IOS 命令序列
        :param actions: 操作列表，每个操作为字典
        """
        commands = []
        for action in actions:
            cmd_type = action.get('type')
            if cmd_type == 'interface':
                commands.append(f"interface {action['name']}")
                if 'description' in action:
                    commands.append(
                        f" description {action['description']}")
                if 'ip_address' in action:
                    ip, mask = action['ip_address'].split('/')
                    # 转换 CIDR 为点分十进制掩码
                    mask = self._cidr_to_mask(int(mask))
                    commands.append(f" ip address {ip} {mask}")
                if 'shutdown' not in action or not action['shutdown']:
                    commands.append(" no shutdown")
                commands.append("exit")

            elif cmd_type == 'vlan':
                commands.append(f"vlan {action['id']}")
                if 'name' in action:
                    commands.append(f" name {action['name']}")
                commands.append("exit")

            elif cmd_type == 'static_route':
                commands.append(
                    f"ip route {action['dest']} "
                    f"{self._cidr_to_mask(action.get('prefix', 24))} "
                    f"{action['next_hop']}"
                )

            elif cmd_type == 'raw':
                commands.append(action['command'])

        return commands

    @staticmethod
    def _cidr_to_mask(prefix: int) -> str:
        """CIDR 前缀转点分十进制掩码"""
        mask = (0xFFFFFFFF << (32 - prefix)) & 0xFFFFFFFF
        return '.'.join(
            str((mask >> (8 * i)) & 0xFF) for i in range(3, -1, -1)
        )

九、总结

思考题

1. 宏（Macro）和模板继承（Block）在配置模板中各自适用于什么场景？请举例说明何时用宏、何时用继承。
2. 如果一个模板中有嵌套的三层for循环（设备→VLAN→接口），性能会如何？如何优化？
3. 设计一个BGP多跳邻居模板：支持配置GR、Add-Path、Route-Reflector等不同BGP特性的组合，且每个特性的开关由变量控制。

下篇预告

第307篇：YAML与JSON——结构化数据格式。学习两种网络自动化中最常用的数据交换格式，掌握设备清单、配置变量、采集结果的存储规范。