python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python自动化巡检

一文分享Python自动化巡检脚本

作者:刘子毅

文章详细介绍了自动化巡检脚本的开发过程,从设备清单、巡检命令、并线多线程并发巡检到生成Excel报告,并推荐了Netmiko、NAPALM、ntc-templates、TextFSM、Paramiko等五个网络自动化工具,可以帮助工程师提高工作效率

一、自动化巡检脚本:从 3 小时到 3 分钟

我们需要什么

一个脚本,能做到:

  1. 读取设备清单(IP、厂商、登录凭据)
  2. 逐台登录,执行预设的巡检命令
  3. 收集结果,按设备整理
  4. 生成一份 Excel 报告,异常项自动标红

技术栈:Netmiko + 多线程 + openpyxl

第一步:设备清单

创建一个 Python 文件,把设备信息写进去。设备少可直接硬编码,多了可换成 CSV 读取。

devices = [
    {"host": "10.1.1.1", "device_type": "cisco_ios"},
    {"host": "10.1.1.2", "device_type": "huawei"},
    {"host": "10.1.2.1", "device_type": "cisco_ios"},
    # 把你的设备全部加进来
]

CREDENTIALS = {
    "username": "admin",
    "password": "your_password",
}

第二步:定义巡检命令

不同厂商命令不一样,用一个字典来做映射。你可以根据环境自由增减。

CHECK_COMMANDS = {
    "cisco_ios": [
        ("show version | include Version", "软件版本"),
        ("show ip interface brief", "接口状态"),
        ("show processes cpu sorted | exclude 0.00%", "CPU占用"),
        ("show memory statistics", "内存使用"),
        ("show logging | include %", "异常日志"),
        ("show ip route summary", "路由表概要"),
    ],
    "huawei": [
        ("display version | include VRP", "软件版本"),
        ("display ip interface brief", "接口状态"),
        ("display cpu-usage", "CPU占用"),
        ("display memory-usage", "内存使用"),
        ("display logbuffer | include Error", "异常日志"),
        ("display ip routing-table statistics", "路由表概要"),
    ],
}

第三步:并发巡检(核心)

一台一台串行太慢,用线程池并发。30 台设备,并发 10 线程,3 分钟全部跑完。

from concurrent.futures import ThreadPoolExecutor, as_completed
from netmiko import ConnectHandler
from datetime import datetime

MAX_WORKERS = 10  # 同时连接 10 台设备

def inspect_device(dev):
    """巡检单台设备,返回结果字典"""
    result = {
        "设备": dev["host"],
        "厂商": dev["device_type"],
        "巡检时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "状态": "成功",
    }

    # 合并凭据
    login = {**dev, **CREDENTIALS}

    try:
        with ConnectHandler(**login) as conn:
            commands = CHECK_COMMANDS.get(dev["device_type"], [])
            for cmd, name in commands:
                try:
                    output = conn.send_command(cmd, read_timeout=20)
                    result[name] = output.strip()
                except Exception as e:
                    result[name] = f"[执行失败] {e}"
    except Exception as e:
        result["状态"] = "连接失败"
        result["错误信息"] = str(e)

    return result

# 并发执行
reports = []
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
    futures = {pool.submit(inspect_device, d): d for d in devices}
    for f in as_completed(futures):
        reports.append(f.result())

print(f"巡检完成:共 {len(reports)} 台设备")

第四步:生成 Excel 报告

拿到数据只是第一步,还得让老板看得懂。生成带颜色标注的 Excel,异常项自动标红。

import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment

wb = openpyxl.Workbook()
ws = wb.active
ws.title = f"设备巡检报告_{datetime.now():%Y%m%d}"

# 表头
headers = ["设备", "厂商", "状态", "软件版本", "接口状态",
           "CPU占用", "内存使用", "异常日志", "路由表概要"]
header_fill = PatternFill(start_color="1F4E79", end_color="1F4E79",
                          fill_type="solid")
header_font = Font(color="FFFFFF", bold=True, size=11)

for col, h in enumerate(headers, 1):
    cell = ws.cell(row=1, column=col, value=h)
    cell.fill = header_fill
    cell.font = header_font
    cell.alignment = Alignment(horizontal="center")

# 异常标记
red_fill = PatternFill(start_color="FFC7CE", end_color="FFC7CE",
                       fill_type="solid")
red_font = Font(color="9C0006")

# 写入数据
for row, report in enumerate(reports, 2):
    for col, key in enumerate(headers, 1):
        value = report.get(key, "")
        ws.cell(row=row, column=col, value=value)

        # 包含异常关键词则标红
        if isinstance(value, str) and any(
            kw in value.lower()
            for kw in ["down", "error", "fail", "失败"]
        ):
            ws.cell(row=row, column=col).fill = red_fill
            ws.cell(row=row, column=col).font = red_font

# 调整列宽
for col in ws.columns:
    ws.column_dimensions[col[0].column_letter].width = 22

filename = f"巡检报告_{datetime.now():%Y%m%d}.xlsx"
wb.save(filename)

运行效果

巡检完成:共 30 台设备
报告已生成:巡检报告_20260521.xlsx

打开 Excel,所有设备一目了然。有异常的自动标红,发给领导比你手动填的表格专业十倍。

进阶方向

这个脚本只是一个起点。你可以在此基础上:

二、会用库的网工,一天干别人一周的活

很多网工学 Python 的瓶颈不在语法,在“不知道有哪些好用的库”。网络自动化领域已经有一套成熟的工具链,这篇帮你一次性搞清楚。

第一把刀:Netmiko —— 网络设备的瑞士军刀

如果你只能学一个库,就学 Netmiko。它是 Paramiko 的上层封装,专门为网络设备做 SSH 连接优化,支持思科、华为、H3C、Juniper 等几十个厂商。

from netmiko import ConnectHandler

dev = {
    "device_type": "huawei_vrp",
    "host": "10.1.1.1",
    "username": "admin",
    "password": "pass",
}

try:
    with ConnectHandler(**dev) as conn:
        print("===== 设备版本信息 =====")
        print(conn.send_command("display version"))
except Exception as e:
    print("连接失败:", e)

Netmiko 帮你自动处理厂商差异——思科 enable 模式,华为 system-view,它都自动完成。

适用场景:日常登录设备执行命令、配置备份、批量巡检。

第二把刀:NAPALM —— 厂商无关的网络抽象

Netmiko 的局限是命令厂商相关。思科 show ip interface brief,华为 display ip interface brief。跨厂商脚本要写一堆 if-else。NAPALM 提供统一 API,换厂商不改代码逻辑。

from napalm import get_network_driver

driver = get_network_driver("ios")
device_ip = "10.1.1.1"
username = "admin"
password = "pass"

try:
    with driver(device_ip, username, password, timeout=30) as dev:
        facts = dev.get_facts()
        print("主机名:", facts["hostname"])
        print("型号:", facts["model"])
        print("版本:", facts["os_version"])

        interfaces = dev.get_interfaces()
        print("\n接口数量:", len(interfaces))

        dev.discard_config()
except Exception as e:
    print("连接/执行失败:", str(e))

杀手功能 compare_config():自动比对当前配置和待下发配置,只显示差异,变更窗口神器。

适用场景:多厂商环境、配置比对、BGP/OSPF 状态监控。

第三把刀:ntc-templates —— 把 show 输出变成结构化数据

我们每天看这样的输出:

Interface              IP-Address      OK? Method Status
GigabitEthernet0/0     10.1.1.1        YES NVRAM  up
GigabitEthernet0/1     unassigned      YES NVRAM  down

想在脚本里拿到“哪个接口是 down 的”,以前得写正则解析。ntc-templates 说:不用了。

from netmiko import ConnectHandler

device = {
    "device_type": "cisco_ios",
    "host": "10.1.1.1",
    "username": "admin",
    "password": "pass"
}

with ConnectHandler(**device) as conn:
    output = conn.send_command("show ip interface brief", use_textfsm=True)

for intf in output:
    print(intf["intf"], intf["status"])

use_textfsm=True 背后就是 ntc-templates 在干活,内置上百个厂商命令解析模板,输出直接变成字典列表。
适用场景:任何需要结构化解析 CLI 输出的场景。

第四把刀:TextFSM —— 自定义解析,适配非标设备

ntc-templates 覆盖不了的非标输出、工控设备、安全设备,用 TextFSM 自己写模板。

模板文件 show_int.template

Value INTF (\S+)
Value IPADDR (\S+)
Value STATUS (up|down)
Value PROTO (up|down)

Start
  ^${INTF}\s+${IPADDR}\s+${STATUS}\s+${PROTO} -> Record

解析代码:

import textfsm

output = """Vlan1      192.168.1.1   up   up
Gigabit0/1  unassigned    down down"""

with open("show_int.template") as tmpl:
    parser = textfsm.TextFSM(tmpl)

result = parser.ParseText(output)
print(result)

写一次模板,永久复用。

适用场景:非标设备、特殊命令输出、工控/安全设备。

第五把刀:Paramiko —— 最灵活的底层

前面四个库底层都是 Paramiko。一般情况下不需要直接用,但跳板机、交互式会话、特殊认证非它不可。

import paramiko
import time

ip = "10.1.1.1"
username = "admin"
password = "pass"

try:
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    ssh.connect(hostname=ip, username=username, password=password, port=22, timeout=10)

    # 华为等设备必须用交互式 shell
    shell = ssh.invoke_shell()
    time.sleep(1)

    shell.send("display version\n")
    time.sleep(2)

    output = shell.recv(65535).decode("utf-8")
    print(output)

    ssh.close()
except Exception as e:
    print(f"连接失败:{str(e)}")

适用场景:跳板机环境、复杂交互、密钥+密码双因素认证。

选型速查

你的需求用哪个库
刚入门,想连设备Netmiko
多厂商,想统一接口NAPALM
想看结构化数据ntc-templates + Netmiko
非标设备,需自定义解析TextFSM
跳板机 / 特殊认证Paramiko

大部分场景,Netmiko + ntc-templates 就够了。等你遇到多厂商环境再加 NAPALM。

总结

手工巡检和一个个敲命令的日子该翻篇了。一个脚本加上这几把“刀”,能把你从重复劳动里解放出来,让你有更多时间钻研架构和优化。如果你的巡检脚本跑起来了,或者还有特殊场景不知道怎么适配,欢迎留言交流。转发给还在手工巡检的同事,大家一起提效。

到此这篇关于一文分享Python自动化巡检脚本的文章就介绍到这了,更多相关Python自动化巡检内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文