第307篇：YAML/JSON 配置数据格式

关键词

YAML、JSON、数据格式、结构化数据、序列化、配置管理、数据驱动、自动化数据定义

一、为什么需要数据格式

1.1 从硬编码到数据驱动

自动化演进的三个阶段：

  L1：硬编码脚本
  ┌─ 所有值写在 Python 代码里
  ├─ 改一个 IP 就要改代码
  └─ 脚本不可复用

  L2：配置文件驱动
  ┌─ 数据写在 YAML/JSON 文件中
  ├─ Python 代码只负责逻辑
  ├─ 改数据不改代码
  └─ 脚本可复用于不同场景

  L3：数据库/CMDB 驱动
  ┌─ 数据存储在 CMDB/数据库中
  ├─ 自动化流程自动读取
  ├─ 实时更新
  └─ 全流程自动化

  JSON 和 YAML 就是 L2 阶段的核心数据载体

1.2 JSON vs YAML 对比

特性 JSON YAML
语法 可读性 注释 多行字符串 类型 解析速度 Python 集成 典型用途 Ansible 严格 一般 ❌ 不支持 ❌ 不方便 自动 快 标准库 json API 数据交换 部分支持 简洁 优秀 ✅ 支持 # ✅ 支持

选择建议： ┌─ 写配置文件/设备数据 → YAML（更可读） ├─ API 交互/数据交换 → JSON（更快、通用） └─ Ansible 项目 → YAML（默认格式）

二、JSON 基础

2.1 JSON 语法

JSON 支持的四种基本结构：

  1. 对象（字典）
  {
    "name": "ACC-SW01",
    "role": "access",
    "location": "B1-Floor01"
  }

  2. 数组（列表）
  ["ACC-SW01", "ACC-SW02", "ACC-SW03"]

  3. 基本类型
  ┌─ 字符串："hello"
  ├─ 数字：100, 3.14
  ├─ 布尔：true, false
  └─ 空值：null

  4. 嵌套
  {
    "device": {
      "name": "ACC-SW01",
      "interfaces": [
        {"name": "GE0/0/1", "ip": "10.0.1.1"}
      ]
    }
  }

  JSON 规则：
  ┌─ 键必须用双引号
  ├─ 字符串必须用双引号
  ├─ 不能有注释
  └─ 最后一个元素不能有逗号

2.2 Python 操作 JSON

#!/usr/bin/env python3
# json_basic.py — JSON 读写操作

import json

# ===== 写入 JSON =====
device = {
    "hostname": "CORE-RT01",
    "vendor": "Huawei",
    "model": "NE20E-S4",
    "os_version": "V800R022C00",
    "interfaces": [
        {
            "name": "GigabitEthernet0/0/1",
            "ip": "10.0.12.1",
            "mask": 30,
            "status": "up",
        },
        {
            "name": "GigabitEthernet0/0/2",
            "ip": "10.0.13.1",
            "mask": 30,
            "status": "up",
        },
    ],
    "vlans": [10, 20, 30],
    "features": {
        "ospf": True,
        "bgp": True,
        "mpls": True,
        "segment_routing": False,
    },
}

# 写入文件（格式化输出）
with open("device.json", "w", encoding="utf-8") as f:
    json.dump(device, f, indent=2, ensure_ascii=False)
print("✓ 已写入 device.json")

# 写入文件（压缩、省空间）
with open("device_min.json", "w") as f:
    json.dump(device, f, separators=(",", ":"))
print("✓ 已写入 device_min.json")

# ===== 读取 JSON =====
with open("device.json", "r", encoding="utf-8") as f:
    loaded = json.load(f)

print(f"\n设备名称: {loaded['hostname']}")
print(f"接口数量: {len(loaded['interfaces'])}")
print(f"启用特性: {[k for k, v in loaded['features'].items() if v]}")

# ===== JSON 转字符串 =====
json_str = json.dumps(device, indent=2, ensure_ascii=False)
print(f"\nJSON 字符串（前 200 字符）:")
print(json_str[:200])

# ===== 字符串转 JSON =====
json_input = '{"name": "SW01", "vlans": [10, 20]}'
parsed = json.loads(json_input)
print(f"\n解析结果: {parsed}")
print(f"名称: {parsed['name']}, VLAN 数: {len(parsed['vlans'])}")

# ===== 网络设备批量数据 =====
devices = [
    {"hostname": "CORE-SW01", "role": "core", "mgmt_ip": "10.0.0.1"},
    {"hostname": "CORE-SW02", "role": "core", "mgmt_ip": "10.0.0.2"},
    {"hostname": "ACC-SW01", "role": "access", "mgmt_ip": "10.0.1.1"},
    {"hostname": "ACC-SW02", "role": "access", "mgmt_ip": "10.0.1.2"},
]

# 写入数组
with open("devices.json", "w") as f:
    json.dump(devices, f, indent=2)

# 只读 access 交换机
access_sw = [d for d in devices if d["role"] == "access"]
print(f"\n接入交换机: {[d['hostname'] for d in access_sw]}")

2.3 JSON 在网络自动化中的典型应用

{
  "_comment": "网络设备清单（设备信息表）",
  "devices": [
    {
      "hostname": "CORE-RT01",
      "mgmt_ip": "10.255.1.1",
      "device_type": "huawei",
      "role": "core-router",
      "location": "DC-A-R01",
      "vendor": "Huawei",
      "model": "NE20E-S4"
    },
    {
      "hostname": "CORE-SW01",
      "mgmt_ip": "10.255.2.1",
      "device_type": "huawei",
      "role": "core-switch",
      "location": "DC-A-R02",
      "vendor": "Huawei",
      "model": "CE12808"
    }
  ]
}

{
  "_comment": "配置参数（VLAN/接口规划）",
  "global": {
    "ntp_server": "203.0.113.1",
    "domain": "corp.local",
    "snmp_community": "public-ro"
  },
  "vlans": [
    {"id": 10, "name": "MGMT", "subnet": "10.0.10.0/24"},
    {"id": 20, "name": "OFFICE", "subnet": "10.0.20.0/24"},
    {"id": 100, "name": "SERVER", "subnet": "10.0.100.0/24"}
  ],
  "interfaces": [
    {"device": "CORE-SW01", "name": "GE0/0/1",
     "desc": "UPLINK-TO-RT01", "mode": "trunk",
     "vlans": "10 20 100"},
    {"device": "CORE-SW01", "name": "GE0/0/2",
     "desc": "ACCESS-FLOOR1", "mode": "access", "vlan": 20}
  ]
}

三、YAML 基础

3.1 YAML 语法

YAML 核心语法：

  # 注释以 # 开头

  # 键值对（字典）
  hostname: CORE-RT01
  role: core-router
  location: "DC-A-R01"   # 引号可选

  # 数组（列表）
  vlans:
    - 10
    - 20
    - 30

  # 或者行内格式
  vlans: [10, 20, 30]

  # 嵌套结构
  interfaces:
    - name: GE0/0/1
      ip: 10.0.12.1
      mask: 30
      status: up
    - name: GE0/0/2
      ip: 10.0.13.1
      mask: 30
      status: up

  # 多行字符串（保留换行）
  config: |
    interface GE0/0/1
     ip address 10.0.1.1 255.255.255.0
     undo shutdown

  # 多行字符串（折叠空格）
  description: >
    This is a long description
    that will be folded into
    a single line.

  # 布尔值
  ospf_enabled: true
  bgp_enabled: false
  debug_mode: yes    # 等同于 true
  verbose_mode: no   # 等同于 false

  YAML 规则：
  ┌─ 缩进用空格（不能用 Tab）
  ├─ 缩进层级表示嵌套
  ├─ - 表示列表元素
  ├─ : 后面必须有空格
  └─ # 开头是注释

3.2 Python 操作 YAML

#!/usr/bin/env python3
# yaml_basic.py — YAML 读写操作

# 安装：python -m pip install pyyaml

import yaml
import json

# ===== YAML 数据（直接写在代码中） =====
yaml_data = """
# 网络设备清单
devices:
  - hostname: CORE-RT01
    vendor: Huawei
    model: NE20E-S4
    mgmt_ip: 10.255.1.1
    role: core
    interfaces:
      - name: GigabitEthernet0/0/0
        ip: 10.0.0.1
        mask: 30
        description: "Uplink to Core-SW"
      - name: GigabitEthernet0/0/1
        ip: 10.0.1.1
        mask: 24
        description: "Access network"

  - hostname: ACC-SW01
    vendor: Huawei
    model: S5735-L48T4XE
    mgmt_ip: 10.255.2.1
    role: access
    vlans: [10, 20, 30, 100]
    interfaces:
      - name: GigabitEthernet0/0/1
        mode: trunk
        allowed_vlans: "10 20 30 100"
      - name: GigabitEthernet0/0/2
        mode: access
        vlan: 10
      - name: GigabitEthernet0/0/3
        mode: access
        vlan: 20

# 全局参数
global:
  ntp_server: 203.0.113.1
  syslog_server: 10.0.0.100
  snmp:
    community: public
    version: v2c
    location: "DC-A-R01"
"""

# ===== 解析 YAML =====
data = yaml.safe_load(yaml_data)
print("=== YAML 解析结果 ===")
print(f"设备数量: {len(data['devices'])}")
for dev in data["devices"]:
    print(f"\n设备: {dev['hostname']}")
    print(f"  厂商: {dev['vendor']} {dev['model']}")
    print(f"  管理 IP: {dev['mgmt_ip']}")
    print(f"  接口数: {len(dev.get('interfaces', []))}")
    print(f"  VLAN: {dev.get('vlans', 'N/A')}")

print(f"\n全局 NTP: {data['global']['ntp_server']}")
print(f"SNMP 版本: {data['global']['snmp']['version']}")

# ===== 写入 YAML =====
with open("network.yml", "w", encoding="utf-8") as f:
    yaml.dump(data, f, default_flow_style=False, allow_unicode=True)
print("\n✓ 已写入 network.yml")

# ===== YAML 与 JSON 互转 =====
json_str = json.dumps(data, indent=2, ensure_ascii=False)
with open("network.json", "w", encoding="utf-8") as f:
    f.write(json_str)
print("✓ 已写入 network.json（YAML → JSON 转换）")

# ===== 从 YAML 文件读取 =====
with open("network.yml", "r", encoding="utf-8") as f:
    loaded = yaml.safe_load(f)

core_devices = [d for d in loaded["devices"] if d["role"] == "core"]
print(f"\n核心设备: {[d['hostname'] for d in core_devices]}")
access_devices = [d for d in loaded["devices"] if d["role"] == "access"]
print(f"接入设备: {[d['hostname'] for d in access_devices]}")

四、实战：数据驱动的配置生成

4.1 完整的数据驱动流程

#!/usr/bin/env python3
# data_driven_config.py — 数据驱动的配置生成

import yaml
import json
from jinja2 import Environment, FileSystemLoader
from netmiko import ConnectHandler
import os

# ===== 1. YAML 数据文件 =====
yaml_device_data = """
# 网络设备数据定义
site_info:
  name: "Corp-Office-B1"
  location: "Building 1, Floor 2"

ntp:
  server: 203.0.113.1
  timezone: "CST+08:00"

devices:
  - hostname: ACC-SW01
    mgmt_ip: 10.0.1.1
    device_type: huawei
    vlans:
      - {id: 10, name: MGMT}
      - {id: 20, name: OFFICE}
      - {id: 30, name: GUEST}
    interfaces:
      - name: GigabitEthernet0/0/1
        description: "UPLINK-TO-CORE"
        mode: trunk
        allowed_vlans: "10 20 30"
      - name: GigabitEthernet0/0/2
        description: "OFFICE-FLOOR1"
        mode: access
        default_vlan: 20
      - name: GigabitEthernet0/0/3
        description: "GUEST-WIFI"
        mode: access
        default_vlan: 30

  - hostname: ACC-SW02
    mgmt_ip: 10.0.1.2
    device_type: huawei
    vlans:
      - {id: 10, name: MGMT}
      - {id: 40, name: PROD}
      - {id: 50, name: DEV}
    interfaces:
      - name: GigabitEthernet0/0/1
        description: "UPLINK-TO-CORE"
        mode: trunk
        allowed_vlans: "10 40 50"
      - name: GigabitEthernet0/0/2
        description: "PROD-SERVER01"
        mode: access
        default_vlan: 40
      - name: GigabitEthernet0/0/3
        description: "DEV-LAB01"
        mode: access
        default_vlan: 50
"""

# ===== 2. 读取数据 =====
data = yaml.safe_load(yaml_device_data)
print(f"站点: {data['site_info']['name']}")
print(f"设备数: {len(data['devices'])}\n")

# ===== 3. 生成 Inventory 数据 =====
inventory = {
    "all": {
        "hosts": {},
        "vars": {
            "ntp_server": data["ntp"]["server"],
            "timezone": data["ntp"]["timezone"],
        },
    }
}

for dev in data["devices"]:
    inventory["all"]["hosts"][dev["hostname"]] = {
        "ansible_host": dev["mgmt_ip"],
        "device_type": dev["device_type"],
        "vlans": dev["vlans"],
        "interfaces": dev["interfaces"],
    }

# 保存为 JSON（给外部工具用）
with open("inventory.json", "w") as f:
    json.dump(inventory, f, indent=2)
print("✓ Inventory 已保存到 inventory.json")

# ===== 4. 生成每个设备的配置 =====
os.makedirs("output_configs", exist_ok=True)
os.makedirs("templates", exist_ok=True)

# 模板
config_template = """\
! ============================================
! 站点: {{ site_name }}
! 设备: {{ device.hostname }}
! 管理 IP: {{ device.mgmt_ip }}
! 生成时间: {{ gen_time }}
! ============================================
#
sysname {{ device.hostname }}
#
{% for vlan in device.vlans %}
vlan {{ vlan.id }}
 name {{ vlan.name }}
#
{% endfor %}
#
{% for iface in device.interfaces %}
interface {{ iface.name }}
 description {{ iface.description }}
 port link-type {{ iface.mode }}
{% if iface.mode == "trunk" %}
 port trunk allow-pass vlan {{ iface.allowed_vlans }}
{% else %}
 port default vlan {{ iface.default_vlan }}
{% endif %}
 undo shutdown
#
{% endfor %}
#
ntp-service unicast-server {{ ntp_server }}
#
return
"""

with open("templates/device_config.j2", "w") as f:
    f.write(config_template)

from datetime import datetime

env = Environment(loader=FileSystemLoader("templates"))
template = env.get_template("device_config.j2")

for dev in data["devices"]:
    output = template.render(
        site_name=data["site_info"]["name"],
        device=dev,
        ntp_server=data["ntp"]["server"],
        gen_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    )

    filename = f"output_configs/{dev['hostname']}_config.txt"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(output)
    print(f"✓ 已生成: {filename} ({len(output.splitlines())} 行)")

# ===== 5. 生成部署脚本 =====
deploy_script = """#!/usr/bin/env python3
# 自动生成的部署脚本

from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoTimeoutException
import yaml

DEVICES = %s

def deploy_device(dev_name, dev_info):
    \"\"\"部署设备配置\"\"\"
    device = {
        "device_type": dev_info["device_type"],
        "host": dev_info["ansible_host"],
        "username": input(f"输入 {dev_name} 用户名: "),
        "password": input(f"输入 {dev_name} 密码: "),
    }

    try:
        conn = ConnectHandler(**device)
    except NetmikoTimeoutException:
        print(f"  [失败] {dev_name} 连接超时")
        return

    # 读取生成的配置
    with open(f"output_configs/{dev_name}_config.txt") as f:
        config = f.read()

    conn.send_config_set(config.split("\\n"))
    conn.save()
    conn.disconnect()
    print(f"  [成功] {dev_name} 配置已下发")

if __name__ == "__main__":
    for name, info in DEVICES.items():
        print(f"部署 {name}...")
        deploy_device(name, info)
"""

# 提取 devices 字典用于部署脚本
deploy_devices = {}
for dev in data["devices"]:
    deploy_devices[dev["hostname"]] = {
        "ansible_host": dev["mgmt_ip"],
        "device_type": dev["device_type"],
    }

with open("deploy_configs.py", "w", encoding="utf-8") as f:
    f.write(deploy_script % json.dumps(deploy_devices, indent=4))

print("\n✓ 部署脚本已生成: deploy_configs.py")
print("\n=== 数据驱动配置生成流程 ===")
print("YAML 数据 → Jinja2 模板 → 配置文件 → 部署脚本")
print("               ↓              ↓              ↓")
print("      人工维护数据    自动渲染生成     自动下发执行")

4.2 从 Excel 生成 YAML 数据

#!/usr/bin/env python3
# excel_to_yaml.py — Excel 转 YAML 配置数据

"""
适用场景：
  网络规划人员在 Excel 中维护设备清单/VLAN 规划
  运维人员通过脚本转为 YAML，再生成配置
"""

import yaml
import json

# 模拟从 Excel 读取的数据
# 实际使用 openpyxl 或 pandas 读取
excel_data = {
    "devices": [
        {
            "hostname": "ACC-SW01",
            "mgmt_ip": "10.0.1.1",
            "role": "access",
            "location": "B1-F1",
            "model": "S5735-L48T4XE",
        },
        {
            "hostname": "ACC-SW02",
            "mgmt_ip": "10.0.1.2",
            "role": "access",
            "location": "B1-F2",
            "model": "S5735-L48T4XE",
        },
        {
            "hostname": "CORE-SW01",
            "mgmt_ip": "10.0.0.1",
            "role": "core",
            "location": "DC-A",
            "model": "CE12808",
        },
    ],
    "vlans": [
        {"id": 10, "name": "MGMT", "subnet": "10.0.10.0/24"},
        {"id": 20, "name": "DATA", "subnet": "10.0.20.0/24"},
        {"id": 100, "name": "SERVER", "subnet": "10.0.100.0/24"},
    ],
    "global": {
        "ntp": "203.0.113.1",
        "snmp_location": "Beijing DC",
    },
}

# 转换为 YAML
with open("network_data.yml", "w", encoding="utf-8") as f:
    yaml.dump(excel_data, f, default_flow_style=False, allow_unicode=True)
print("✓ Excel 数据已转换为 network_data.yml")

# 验证
with open("network_data.yml", "r") as f:
    loaded = yaml.safe_load(f)

print(f"设备数: {len(loaded['devices'])}")
print(f"VLAN 数: {len(loaded['vlans'])}")
print(f"NTP: {loaded['global']['ntp']}")

五、数据验证

5.1 数据完整性检查

#!/usr/bin/env python3
# validate_data.py — 数据验证

import yaml
import json
import sys

def validate_device_data(data):
    """验证设备数据的完整性"""
    errors = []
    warnings = []

    if "devices" not in data:
        errors.append("缺少 devices 字段")

    for i, dev in enumerate(data.get("devices", [])):
        prefix = f"设备 #{i+1}"

        # 必填字段
        for field in ["hostname", "mgmt_ip"]:
            if field not in dev:
                errors.append(f"{prefix}: 缺少 {field}")

        # IP 格式检查
        if "mgmt_ip" in dev:
            parts = dev["mgmt_ip"].split(".")
            if len(parts) != 4 or not all(
                p.isdigit() and 0 <= int(p) <= 255 for p in parts
            ):
                errors.append(f"{prefix}: mgmt_ip 格式错误 {dev['mgmt_ip']}")

        # 主机名校验
        if "hostname" in dev:
            if not dev["hostname"].replace("-", "").isalnum():
                warnings.append(
                    f"{prefix}: hostname 包含特殊字符 {dev['hostname']}"
                )

        # 接口校验
        for j, iface in enumerate(dev.get("interfaces", [])):
            iface_prefix = f"{prefix} 接口 #{j+1}"
            if "mode" in iface and iface["mode"] not in ["access", "trunk", "hybrid"]:
                errors.append(f"{iface_prefix}: 未知 mode '{iface['mode']}'")

    return errors, warnings

# 测试
test_data = yaml.safe_load("""
devices:
  - hostname: "ACC-SW01"
    mgmt_ip: "10.0.1.1"
    interfaces:
      - name: GE0/0/1
        mode: trunk
  - hostname: "INVALID_IP_SW"
    mgmt_ip: "10.0."     # 错误 IP
    interfaces:
      - name: GE0/0/1
        mode: invalid_mode   # 错误模式
  - hostname: ""           # 空主机名
    mgmt_ip: "10.0.2.1"
""")

errors, warnings = validate_device_data(test_data)
if errors:
    print("=== 错误 ===")
    for e in errors:
        print(f"  ✗ {e}")
if warnings:
    print("=== 警告 ===")
    for w in warnings:
        print(f"  △ {w}")
if not errors and not warnings:
    print("✓ 数据校验通过")

六、最佳实践

6.1 数据组织规范

推荐的数据组织方式：

  1. 分层次
  ┌─ site: 站点/项目信息
  ├─ global: 全局配置参数
  ├─ devices: 设备清单
  ├─ vlans: VLAN 规划
  ├─ subnets: IP 地址规划
  └─ links: 链路信息

  2. 独立文件
  ┌─ site.yml — 站点信息
  ├─ devices.yml — 设备清单
  ├─ vlans.yml — VLAN 规划
  ├─ network.yml — 网络参数
  └─ secrets.yml — 敏感信息（加密）

  3. 版本管理
  ┌─ 所有数据文件纳入 Git 管理
  ├─ 每次变更 commit + PR review
  ├─ 标签对应发布版本
  └─ 回滚历史版本

  4. 数据校验
  ┌─ 提交前自动校验
  ├─ 定义 JSON Schema
  ├─ 检查必填字段
  └─ IP 格式/唯一性检查

6.2 敏感信息处理

#!/usr/bin/env python3
# secrets_handling.py — 凭证管理

"""
安全准则：
  1. 永远不要将密码硬编码在 YAML/JSON 中
  2. 使用环境变量或专用密钥管理工具
  3. 使用 ansible-vault 加密敏感数据
"""

import os
import yaml

# 方式 1：环境变量
username = os.environ.get("NET_USER", "")
password = os.environ.get("NET_PASSWORD", "")

if not username or not password:
    # 交互式输入
    import getpass
    username = input("用户名: ")
    password = getpass.getpass("密码: ")

# 方式 2：独立的加密凭证文件
# secrets.yml（不上传到 Git）
# username: admin
# password: YourRealPassword123

# 在 .gitignore 中添加 secrets.yml

# 方式 3：使用 keyring 系统
try:
    import keyring
    password = keyring.get_password("network_automation", username)
except ImportError:
    pass

# 方式 4：Ansible Vault 加密
# ansible-vault encrypt secrets.yml
# ansible-vault decrypt secrets.yml

七、总结

JSON 和 YAML 在网络自动化中的角色：

JSON — 程序间数据交换 ┌─ API 请求/响应 ├─ 数据库存储 ├─ 配置文件备份 └─ Python/JavaScript 互操作

YAML — 人类可读的配置 ┌─ Ansible Playbook ├─ 设备数据定义 ├─ 网络规划文档 └─ 自动化输入数据

数据驱动自动化工作流： | YAML数据 定义 | → | Jinja2 渲染 | → | Netmiko 下发 | → | 设备配置 生效 | | --- | --- | --- | --- | --- | --- | --- | 人工维护 自动生成 自动执行 验证确认

下篇预告：第308篇 — Ansible Network Modules 实战，将深入 Ansible 的网络自动化模块，介绍如何通过 Playbook 批量管理网络设备。