🚀 Nornir 网络自动化实战实验手册

01 为什么要写这本手册：从"人肉运维"到"自动化躺赢"

想象一下：凌晨两点，你的老板打电话来说——"兄弟，把全网 200 台交换机的 NTP 服务器改一下，顺便检查一下 50 台 Linux 服务器的磁盘。"如果你还在用 SSH 一台一台登，那这辈子算是交给机房了。😱

这就是为什么我们需要 Nornir。它是用 Python 写的网络自动化框架，但它不是一个"让你学新语法"的工具——它就是把你本来要手敲的命令，用一个优雅的、并行的、可复用的方式，批量甩给一堆设备。它的设计哲学是："Inventory（清单）+ Tasks（任务）+ Results（结果）"，三层结构清晰得像三明治。

1.1 Nornir 与 Ansible 的"爱恨情仇"

结论：如果你已经是 Pythonista，Nornir 会让你爽到飞起；如果你只想快速上手、团队又不写代码，Ansible 也挺好。但本手册的目标，就是让你成为那个"既懂网络又会写代码"的稀缺物种。💪

1.2 本手册的实验拓扑

为了让实验贴近真实，我们设计了一个"小但完整"的实验室环境。你可以用真机、EVE-NG、GNS3、CML 或容器来复现，IP 和用户名都可按需修改。

02 环境准备与安装：搭好你的"自动化指挥所"

2.1 Python 虚拟环境（强烈推荐）

为什么要用虚拟环境？因为你的电脑里可能装着十个项目，每个项目依赖的库版本都不一样。虚拟环境就像给每个项目一个独立的"沙盒房间"，互不干扰。

# 创建项目目录
mkdir ~/nornir-lab && cd ~/nornir-lab

# 创建虚拟环境（Python 3.9+）
python3 -m venv venv

# 激活（Linux/Mac）
source venv/bin/activate
# Windows: venv\Scripts\activate

# 升级 pip
pip install --upgrade pip

2.2 安装 Nornir 与插件全家桶

Nornir 采用"核心 + 插件"的架构。核心很轻，功能靠插件扩展，就像乐高积木。

# 核心
pip install nornir

# 连接插件：Netmiko（SSH，对 Cisco 超友好）
pip install nornir-netmiko

# 连接插件：Napalm（多厂商统一抽象层）
pip install nornir-napalm

# 实用插件：表格美化输出
pip install nornir-utils

# 配置管理插件（可选，Jinja2 渲染）
pip install nornir-jinja2

# 给 Linux 用的 Paramiko SSH
pip install paramiko

nornir>=3.3.0
nornir-netmiko>=0.4.0
nornir-napalm>=0.5.0
nornir-utils>=0.2.0
nornir-jinja2>=0.3.0
paramiko>=3.0.0

2.3 验证安装

python -c "import nornir; print(nornir.__version__)"

如果能看到版本号（如 3.3.0），恭喜，你的自动化指挥所已经搭好了！🎉

03 Inventory（清单）：告诉 Nornir "你有哪些兵"

Inventory 是 Nornir 的"花名册"。它记录每台设备的 IP、平台、用户名、密码、分组等信息。Nornir 支持多种来源：SimpleInventory（YAML 文件）、AnsibleInventory、NetBoxInventory、自定义 Python 类等。新手从 YAML 起步最省心。

3.1 目录结构规划

nornir-lab/
├── config.yaml          # Nornir 主配置（指定 inventory 来源）
├── inventory/
│   ├── hosts.yaml       # 主机清单
│   ├── groups.yaml      # 分组与组级默认参数
│   └── defaults.yaml    # 全局默认值
├── secrets.yaml         # 敏感信息（生产环境请用环境变量 / Vault）
├── tasks/               # 任务脚本目录
│   ├── 01_hello.py
│   ├── 02_backup.py
│   └── ...
└── templates/           # Jinja2 配置模板
    └── cisco_base.j2

3.2 主配置文件 config.yaml

inventory:
  plugin: SimpleInventory
  options:
    host_file: inventory/hosts.yaml
    group_file: inventory/groups.yaml
    defaults_file: inventory/defaults.yaml

runner:
  plugin: ThreadedRunner
  options:
    num_workers: 10   # 并发线程数，按设备数量调整

logging:
  enabled: true
  level: INFO
  log_file: nornir.log

3.3 主机清单 hosts.yaml

# ---- Cisco 网络设备 ----
core-sw:
  hostname: 10.0.0.1
  groups: [cisco, core]
  data:
    role: core
    mgmt_ip: 10.0.0.1

dist-sw-01:
  hostname: 10.0.0.11
  groups: [cisco, distribution]
  data:
    role: distribution
    mgmt_ip: 10.0.0.11

dist-sw-02:
  hostname: 10.0.0.12
  groups: [cisco, distribution]
  data:
    role: distribution
    mgmt_ip: 10.0.0.12

# ---- Linux 服务器 ----
linux-srv-1:
  hostname: 10.0.0.21
  groups: [linux, ubuntu]
  platform: linux
  data:
    os: ubuntu
    role: web

linux-srv-2:
  hostname: 10.0.0.22
  groups: [linux, rhel]
  platform: linux
  data:
    os: rocky
    role: db

linux-srv-3:
  hostname: 10.0.0.23
  groups: [linux, debian]
  platform: linux
  data:
    os: debian
    role: cache

3.4 分组配置 groups.yaml

分组是 Nornir 的杀手特性——把设备按"角色/厂商/位置"分组，任务就能针对某个组执行。组级参数会继承给组内所有主机。

cisco:
  platform: ios
  connection_options:
    netmiko:
      extras:
        device_type: cisco_ios
        secret: "cisco"   # enable 密码
        global_delay_factor: 2

distribution:
  data:
    vlan_range: "10-50"

linux:
  platform: linux
  connection_options:
    paramiko:
      extras:
        look_for_keys: false
        allow_agent: false

3.5 全局默认值 defaults.yaml

username: admin
password: "Cisco123!"   # 实验室用，生产请用环境变量
port: 22
timeout: 30

import os
from nornir import InitNornir

nr = InitNornir(
    config_file="config.yaml",
    inventory={
        "options": {
            "defaults_file": {
                "username": os.getenv("NORNIR_USER", "admin"),
                "password": os.getenv("NORNIR_PASS"),
            }
        }
    },
)

3.6 初始化 Nornir & 查看清单

创建 tasks/00_init.py：

from nornir import InitNornir

nr = InitNornir(config_file="config.yaml")

# 打印所有主机
print("📋 我的设备清单：")
for name, host in nr.inventory.hosts.items():
    print(f"  - {name:15s} {host.hostname:12s} 组={host.groups}")

# 按组筛选
cisco_devs = nr.filter(platform="ios")
linux_devs = nr.filter(platform="linux")
print(f"\nCisco 设备数: {len(cisco_devs.inventory.hosts)}")
print(f"Linux 设备数: {len(linux_devs.inventory.hosts)}")

运行：python tasks/00_init.py，你应该看到整齐的设备列表。如果报错，跳到 第 11 章排错。

04 建立连接：第一次"握手"成功了吗？

Nornir 通过 Connection Plugin 连接设备。对 Cisco，我们用 netmiko（基于 Paramiko 的高级封装，自动处理分页、enable 模式，简直是 Cisco 运维的亲儿子）；对 Linux，直接用 paramiko 或系统 SSH。

4.1 任务函数（Task）的基本结构

Nornir 的任务就是一个接收 task: Task 参数的函数。你在里面写业务逻辑，Nornir 负责把这个函数并行地甩给每台设备。

from nornir import InitNornir
from nornir_netmiko.tasks import netmiko_send_command

nr = InitNornir(config_file="config.yaml")

# 定义任务：查看版本
def get_version(task):
    task.run(
        name="查看设备版本",
        task=netmiko_send_command,
        command_string="show version | include Version",
    )

# 只在 Cisco 设备上执行
result = nr.run(task=get_version)

# 打印结果
print(result)

4.2 运行 & 解读结果

运行后你会看到类似输出：

get_version*****************************************************************
* core-sw ** changed : False **********************************************
vvvv get_version ** changed : False vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv SUCCESS
* dist-sw-01 ** changed : False *******************************************
vvvv get_version ** changed : False vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv SUCCESS
...

关键点：changed 表示是否有配置变更（仅查看=Flase），SUCCESS / FAILED 表示执行状态。Nornir 的结果对象是树形结构，可以精细遍历。

4.3 手动打开/关闭连接（高级控制）

def manual_connection(task):
    # 手动打开 netmiko 连接
    task.host.open_connection(
        connection="netmiko",
        configuration=task.nornir.config,
    )
    conn = task.host.get_connection("netmiko", task.nornir.config)
    output = conn.send_command("show ip interface brief")
    print(f"[{{task.host.name}}]\n{output}")

    # 记得关连接（或用 with 上下文）
    task.host.close_connection("netmiko")

一般不需要手动开关——Nornir 会在任务开始时自动连接、结束时自动关闭。只有当你要在多个 task 之间复用同一个连接时才手动管理。

netmiko_send_command(command_string="show running-config", enable=True)

05 任务（Tasks）基础：学会"发号施令"

5.1 内置任务 vs 自定义任务

Nornir 生态提供了一批现成任务（在 nornir_netmiko.tasks、nornir_napalm.tasks 里），你也可以写自己的。组合复用是王道。

5.2 示例：并发 Ping 检测

from nornir import InitNornir
from nornir_netmiko.tasks import netmiko_send_command

nr = InitNornir(config_file="config.yaml")

def ping_test(task, target):
    # Cisco 用 ping，Linux 后面单独处理
    if task.host.platform == "ios":
        cmd = f"ping {target}"
    else:
        cmd = f"ping -c 4 {target}"

    task.run(
        name=f"Ping {target}",
        task=netmiko_send_command if task.host.platform == "ios" else _linux_run,
        command_string=cmd,
    )

# 对所有设备 ping 网关 10.0.0.254
result = nr.run(task=ping_test, target="10.0.0.254")

5.3 结果处理：漂亮的表格输出

默认打印太朴素？用 nornir_utils 输出成表格、JSON、HTML：

from nornir_utils.plugins.functions import print_result, write_file
from nornir_utils.plugins.task_results import to_json

# 终端彩色表格
print_result(result)

# 导出为 JSON（方便后续处理 / 入库）
write_file("output/ping_result.json", to_json(result))

5.4 分组执行 & 过滤器的艺术

# 只在核心交换机上执行
nr.filter(name="core-sw").run(...)

# 用 group 过滤
nr.filter(filter_func=lambda h: "cisco" in h.groups).run(...)

# 复合条件
nr.filter(platform="ios", role="distribution").run(...)

06 Cisco 实战：把交换机"安排"得明明白白

重头戏来了！本章通过 6 个真实场景，覆盖 Cisco 自动化 90% 的日常需求。

6.1 场景一：批量备份配置文件 💾

备份是自动化的"第一课"，也是"救命课"。每周自动备份 running-config，出问题随时回滚。

# tasks/backup.py
import os
from datetime import datetime
from nornir import InitNornir
from nornir_netmiko.tasks import netmiko_send_command

BACKUP_DIR = "backups"
os.makedirs(BACKUP_DIR, exist_ok=True)

def backup_config(task):
    r = task.run(
        name="抓取 running-config",
        task=netmiko_send_command,
        command_string="show running-config",
        enable=True,
    )
    config = r.result
    date = datetime.now().strftime("%Y%m%d")
    filename = f"{BACKUP_DIR}/{task.host.name}_{date}.cfg"
    with open(filename, "w") as f:
        f.write(config)
    print(f"✅ [{task.host.name}] 已保存: {filename}")

nr = InitNornir(config_file="config.yaml")
nr.filter(platform="ios").run(task=backup_config)

6.2 场景二：批量推送配置（添加 VLAN + NTP）🔧

用 netmiko_send_config 推送配置块，支持"干跑（dry_run）"——先预览再真干，安全！

from nornir_netmiko.tasks import netmiko_send_config

def configure_devices(task, dry_run=False):
    configs = [
        "vlan 100",
        " name USERS",
        "vlan 200",
        " name SERVERS",
        "ntp server 10.0.0.254",
        "ntp source Loopback0",
        "hostname {{ host.name }}",   # 支持 Jinja2 变量！
    ]
    task.run(
        name="推送配置",
        task=netmiko_send_config,
        config_commands=configs,
        dry_run=dry_run,   # True = 只模拟不生效
    )

nr.run(task=configure_devices, dry_run=True)   # 先干跑看效果
# nr.run(task=configure_devices, dry_run=False)  # 确认无误后真干

6.3 场景三：用 Jinja2 模板做"千人千面"的配置

不同设备配置不同，但结构相似——这正是模板的用武之地。创建 templates/cisco_base.j2：

!-- 基础模板：{{ host.name }} --
hostname {{ host.name }}
!
{% if host.data.role == "core" %}
spanning-tree mode rapid-pvst
spanning-tree vlan 1-4094 priority 8192
{% else %}
spanning-tree mode rapid-pvst
{% endif %}
!
{% for vlan_id in range(10, 51) %}
vlan {{ vlan_id }}
{% endfor %}
!
ntp server {{ ntp_server | default("10.0.0.254") }}
logging host {{ syslog_server | default("10.0.0.100") }}
!
interface Loopback0
 ip address {{ host.data.mgmt_ip }} 255.255.255.255
!
end

渲染并推送：

from nornir_jinja2.tasks import template_file
from nornir_netmiko.tasks import netmiko_send_config

def deploy_from_template(task):
    # 1. 渲染模板
    r = task.run(
        name="渲染配置模板",
        task=template_file,
        template="cisco_base.j2",
        path="templates",
        ntp_server="10.0.0.254",
        syslog_server="10.0.0.100",
    )
    config_text = r.result

    # 2. 推送配置
    task.run(
        name="应用配置",
        task=netmiko_send_config,
        config_commands=config_text.splitlines(),
    )

nr.filter(platform="ios").run(task=deploy_from_template)

6.4 场景四：配置变更 + 自动保存 + 差异对比

好习惯：变更前备份 → 推送 → 校验 → 保存。Napalm 还能算配置 diff。

from nornir_napalm.tasks import napalm_configure, napalm_get

def safe_deploy(task):
    # 1. 获取当前配置（用于对比）
    before = task.run(task=napalm_get, getters=["config"])
    running_before = before.result["config"]["running"]

    # 2. 用候选配置替换（Napalm 用 load_replace 语义）
    new_config = "... " # 你的完整配置
    task.run(
        name="应用候选配置",
        task=napalm_configure,
        configuration=new_config,
        replace=True,
    )

    # 3. 再次获取，做差异分析
    after = task.run(task=napalm_get, getters=["config"])
    running_after = after.result["config"]["running"]

    import difflib
    diff = difflib.unified_diff(
        running_before.splitlines(),
        running_after.splitlines(),
        lineterm="",
    )
    print(f"\n📝 [{task.host.name}] 配置差异:")
    print("\n".join(diff))

6.5 场景五：批量采集状态 & 生成报表 📊

from nornir_napalm.tasks import napalm_get
import json

def collect_status(task):
    # Napalm 提供统一 getter，跨厂商通用
    task.run(task=napalm_get, getters=[
        "facts",        # 设备基础信息
        "interfaces",   # 接口状态
        "bgp_neighbors",# BGP 邻居
    ])

result = nr.filter(platform="ios").run(task=collect_status)

# 汇总成 JSON 报表
report = {}
for host_name, host_result in result.items():
    facts = host_result["napalm_get"].result["facts"]
    report[host_name] = {
        "model": facts["model"],
        "os_version": facts["os_version"],
        "uptime": facts["uptime"],
    }

with open("output/device_report.json", "w") as f:
    json.dump(report, f, indent=2)
print(json.dumps(report, indent=2))

6.6 场景六：OS 升级 / 可靠性测试（概念代码）

def upgrade_ios(task, image_path, new_version):
    conn = task.host.get_connection("netmiko", task.nornir.config)

    # 1. 上传镜像（SCP / TFTP，略）
    # conn.send_command(f"copy scp: flash:/{image_path}")

    # 2. 设置 boot 变量
    conn.send_config_set([
        f"boot system flash:/{image_path}",
        "write memory",
    ])

    # 3. 校验（此处省略 md5 比对）

    # 4. reload（危险！建议人工确认）
    # conn.send_command("reload", expect_string="confirm")
    # conn.send_command("\n", expect_string="confirm")

    print(f"⚠️ [{task.host.name}] 已准备升级，等待人工触发 reload")

07 Linux 服务器实战：让 Nornir "跨界"管理服务器

很多人以为 Nornir 只管网络，其实它万物皆可管——只要能通过 SSH 执行命令。Linux 服务器用 Paramiko 连接，配合 shell 命令，运维效率直接起飞。

7.1 自定义 Linux 连接插件思路

Nornir 没有官方的"Linux 专属插件"，但我们可以用 Paramiko Connection Plugin（Nornir 内置支持）直接跑 shell。先写一个复用连接的任务：

# tasks/linux_base.py
from nornir import InitNornir
from nornir.core.task import Result

nr = InitNornir(config_file="config.yaml")

def linux_exec(task, command):
    """通过 paramiko 在 Linux 上执行命令"""
    try:
        conn = task.host.get_connection("paramiko", task.nornir.config)
    except Exception:
        task.host.open_connection("paramiko", configuration=task.nornir.config)
        conn = task.host.get_connection("paramiko", task.nornir.config)

    stdin, stdout, stderr = conn.exec_command(command)
    output = stdout.read().decode("utf-8")
    err = stderr.read().decode("utf-8")
    return Result(host=task.host, result=output, failed=bool(err))

7.2 场景一：采集服务器健康状态 🩺

import re
from nornir.core.task import Result

def health_check(task):
    cmds = {
        "uptime": "uptime",
        "disk": "df -h | grep -v tmpfs",
        "mem": "free -m",
        "cpu_load": "top -bn1 | grep 'load average'",
    }
    report = {}
    for key, cmd in cmds.items():
        r = task.run(task=linux_exec, command=cmd)
        report[key] = r.result.strip()

    # 简单告警逻辑：磁盘使用率 > 80%
    disk_out = report["disk"]
    for line in disk_out.splitlines():
        usage = re.search(r"(\d+)%", line)
        if usage and int(usage.group(1)) > 80:
            print(f"🚨 [{task.host.name}] 磁盘告警: {line}")

    return Result(host=task.host, result=report)

result = nr.filter(platform="linux").run(task=health_check)

7.3 场景二：批量部署监控 Agent（Node Exporter）📦

def deploy_node_exporter(task):
    steps = [
        "curl -sL -o /tmp/node_exporter.tar.gz https://github.com/prometheus/node_exporter/releases/download/v1.6.0/node_exporter-1.6.0.linux-amd64.tar.gz",
        "cd /tmp && tar xzf node_exporter.tar.gz",
        "sudo cp /tmp/node_exporter-*/node_exporter /usr/local/bin/",
        "sudo useradd -rs /bin/false node_exporter || true",
    ]
    for cmd in steps:
        task.run(task=linux_exec, command=cmd)

    # 生成 systemd unit 文件
    unit = """[Unit]
Description=Node Exporter
After=network.target
[Service]
User=node_exporter
ExecStart=/usr/local/bin/node_exporter
[Install]
WantedBy=multi-user.target
"""
    # 写入文件（用 tee 免交互）
    write_cmd = f"echo '{unit}' | sudo tee /etc/systemd/system/node_exporter.service"
    task.run(task=linux_exec, command=write_cmd)
    task.run(task=linux_exec, command="sudo systemctl daemon-reload && sudo systemctl enable --now node_exporter")
    print(f"✅ [{task.host.name}] Node Exporter 部署完成")

nr.filter(platform="linux").run(task=deploy_node_exporter)

7.4 场景三：统一用户 & SSH 密钥管理 👤

def manage_user(task, username="deploy", pubkey=""):
    # 创建用户 + 加 sudo
    task.run(task=linux_exec, command=f"sudo useradd -m -s /bin/bash {username} || true")
    task.run(task=linux_exec, command=f"echo '{username} ALL=(ALL) NOPASSWD:ALL' | sudo tee /etc/sudoers.d/{username}")

    # 部署 SSH 公钥
    if pubkey:
        task.run(task=linux_exec, command=f"sudo mkdir -p /home/{username}/.ssh")
        task.run(task=linux_exec, command=f"echo '{pubkey}' | sudo tee /home/{username}/.ssh/authorized_keys")
        task.run(task=linux_exec, command=f"sudo chmod 600 /home/{username}/.ssh/authorized_keys && sudo chown -R {username}:{username} /home/{username}/.ssh")

    print(f"👤 [{task.host.name}] 用户 {username} 已就绪")

# 从文件读取公钥
with open("~/.ssh/id_ed25519.pub") as f:
    PUBKEY = f.read().strip()

nr.filter(platform="linux").run(task=manage_user, username="deploy", pubkey=PUBKEY)

7.5 场景四：安全基线核查（CIS Benchmark 精简版）🔒

def security_audit(task):
    checks = {
        "SSH root 登录": "grep '^PermitRootLogin' /etc/ssh/sshd_config || echo '未配置'",
        "密码策略": "grep '^PASS_MAX_DAYS' /etc/login.defs || echo '未配置'",
        "防火墙状态": "systemctl is-active firewalld 2>/dev/null || ufw status 2>/dev/null || echo '无防火墙'",
        "异常 SUID": "find / -perm -4000 -type f 2>/dev/null | wc -l",
    }
    audit = {}
    for name, cmd in checks.items():
        r = task.run(task=linux_exec, command=cmd)
        audit[name] = r.result.strip()

    # 输出为结构化结果，方便聚合
    return Result(host=task.host, result=audit)

result = nr.filter(platform="linux").run(task=security_audit)
for host, hr in result.items():
    print(f"\n🔒 {host} 安全核查:")
    for k, v in hr.result.items():
        print(f"   {k}: {v}")

08 批量编排：一个脚本"搞定全网"的终极形态

把前面的技能串起来，写一个"超级编排脚本"——一次执行完成：备份 → 健康检查 → 配置推送 → 校验 → 报告生成。

8.1 编排脚本 tasks/orchestrate.py

import json, os
from datetime import datetime
from nornir import InitNornir
from nornir_netmiko.tasks import netmiko_send_command, netmiko_send_config
from nornir_napalm.tasks import napalm_get
from nornir.core.task import Result

nr = InitNornir(config_file="config.yaml")
os.makedirs("output", exist_ok=True)
LOG = []

def step_backup(task):
    r = task.run(task=netmiko_send_command, command_string="show running-config", enable=True)
    fname = f"output/{task.host.name}_backup.cfg"
    open(fname, "w").write(r.result)
    return Result(host=task.host, result=f"备份 → {fname}")

def step_health(task):
    r = task.run(task=napalm_get, getters=["facts", "interfaces"])
    facts = r.result["facts"]
    return Result(host=task.host, result={
        "model": facts["model"], "version": facts["os_version"],
    })

def step_configure(task):
    cmds = ["ntp server 10.0.0.254", "logging host 10.0.0.100"]
    task.run(task=netmiko_send_config, config_commands=cmds, dry_run=False)
    return Result(host=task.host, result="配置已推送")

def step_verify(task):
    r = task.run(task=netmiko_send_command, command_string="show ntp associations")
    return Result(host=task.host, result=r.result)

# ---- 主流程 ----
print("🚀 开始全网编排...")

cisco = nr.filter(platform="ios")

# Step 1: 备份
r1 = cisco.run(task=step_backup)
# Step 2: 健康检查
r2 = cisco.run(task=step_health)
# Step 3: 配置
r3 = cisco.run(task=step_configure)
# Step 4: 校验
r4 = cisco.run(task=step_verify)

# ---- 生成汇总报告 ----
report = {"timestamp": datetime.now().isoformat(), "devices": {}}
for name in cisco.inventory.hosts:
    report["devices"][name] = {
        "backup": r1[name].failed if name in r1 else True,
        "health": r2[name].result if name in r2 and not r2[name].failed else None,
        "config_ok": not r3[name].failed if name in r3 else False,
        "verify": r4[name].result if name in r4 and not r4[name].failed else "",
    }

with open("output/full_report.json", "w") as f:
    json.dump(report, f, indent=2)

# ---- 打印失败摘要 ----
failures = [n for n, d in report["devices"].items() if d["config_ok"] is False]
if failures:
    print(f"\n❌ 失败设备: {failures}")
else:
    print("\n🎉 全网编排完成，无失败！")

09 配置校验与回滚：把"翻车"概率降到零

9.1 提交前校验（Pre-commit Check）

def pre_check(task):
    # 记录变更前的接口 up 数量
    r = task.run(task=netmiko_send_command, command_string="show ip interface brief | include up")
    up_count = len(r.result.splitlines())
    return Result(host=task.host, result={"up_before": up_count})

def post_check(task, pre_result):
    r = task.run(task=netmiko_send_command, command_string="show ip interface brief | include up")
    up_after = len(r.result.splitlines())
    before = pre_result[task.host.name].result["up_before"]
    if up_after < up_before:
        print(f"🚨 [{task.host.name}] 警告：up 接口数从 {before} 降到 {up_after}！")
    return Result(host=task.host, result={"up_after": up_after})

9.2 自动回滚机制

def deploy_with_rollback(task, configs, pre_checks_ok):
    # 推送配置
    r = task.run(task=netmiko_send_config, config_commands=configs)

    # 校验失败则回滚到备份配置
    if not pre_checks_ok:
        backup = open(f"output/{task.host.name}_backup.cfg").read()
        task.run(task=netmiko_send_config, config_commands=backup.splitlines())
        print(f"⏪ [{task.host.name}] 已自动回滚")
        return Result(host=task.host, failed=True, result="已回滚")
    return Result(host=task.host, result="部署成功")

10 进阶玩法：让自动化"活"起来

10.1 集成 NetBox 作为动态 Inventory

设备越多的，手维护 YAML 越痛苦。用 NetBox（开源 DCIM/IPAM）做唯一数据源，Nornir 直接从 API 拉清单：

pip install nornir-netbox

# config.yaml
inventory:
  plugin: NetboxInventory2
  options:
    nb_url: "https://netbox.example.com"
    nb_token: "{{ env.NETBOX_TOKEN }}"
    filter:
      site: lab

10.2 异步执行（大规模设备加速）

# 使用 ProcessPoolExecutor 应对 >1000 台设备
from concurrent.futures import ProcessPoolExecutor
from nornir import InitNornir

nr = InitNornir(config_file="config.yaml")

def run_on_host(host_name):
    # 每个进程独立 InitNornir（进程内线程池）
    local_nr = InitNornir(config_file="config.yaml")
    host = local_nr.inventory.hosts[host_name]
    # ... 执行任务 ...
    return host_name, "done"

hosts = list(nr.inventory.hosts.keys())
with ProcessPoolExecutor(max_workers=8) as pool:
    results = pool.map(run_on_host, hosts)

10.3 与 pytest 结合做"网络单元测试"

# tests/test_ntp.py
import pytest
from nornir import InitNornir

@pytest.fixture(scope="module")
def nr():
    return InitNornir(config_file="config.yaml")

def test_ntp_configured(nr):
    def check(task):
        r = task.run(task=netmiko_send_command, command_string="show run | include ntp server")
        assert "10.0.0.254" in r.result, f"{task.host.name} NTP 未配置！"
    result = nr.filter(platform="ios").run(task=check)
    assert not result.failed

运行 pytest tests/ -v——把网络验收变成自动化测试，CI/CD 直接接入，这才是真正的 "Network as Code"。🔥

10.4 Webhook 告警集成（企业微信示例）

import requests
def notify_wechat(content):
    webhook = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    requests.post(webhook, json={
        "msgtype": "markdown",
        "markdown": {"content": content}
    })

# 在编排脚本的失败处理里调用
if failures:
    notify_wechat(f"🚨 自动化部署失败设备: {failures}")

11 排错宝典：当自动化"不自动"时怎么办？

11.1 调试技巧：逐步缩小范围

1. 先确认 Inventory 加载正确：print(nr.inventory.hosts)
2. 单台设备跑通：nr.filter(name="core-sw").run(task=...)
3. 开 DEBUG 看原始交互日志
4. 把命令手动 SSH 执行一遍，对比输出

确认 task 函数是幂等的（重复执行不报错）

11.2 常见错误速查表

12 综合实战挑战：检验你的学习成果 🏆

学完了别急着走，试试这几个"毕业设计"级别的任务，把你变成一个真正的网络自动化工程师！

12.1 毕业设计参考架构

📎 附录：常用命令速查 & 资源清单

A. Nornir 常用 API 速查

B. Netmiko 常用 device_type

C. 推荐学习资源

• 📘 Nornir 官方文档：nornir.readthedocs.io
• 📘 Netmiko 文档：netmiko.readthedocs.io
• 📘 Napalm 文档：napalm.readthedocs.io
• 📖 书籍：《Network Automation with Nornir》 by David Barroso
• 🎥 视频：YouTube "Nornir Network Automation" 系列教程
• 💻 GitHub：搜索 nornir-examples 参考实战项目

D. 一份"自动化成熟度"自评表

做完本手册的实验，你应该已经稳稳站在 L3 ~ L4 之间了。继续加油，未来的"网络自动化大神"！🚀