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"},
    # 把你的設(shè)備全部加進(jìn)來(lái)
]

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

CHECK_COMMANDS = {
    "cisco_ios": [
        ("show version | include Version", "軟件版本"),
        ("show ip interface brief", "接口狀態(tài)"),
        ("show processes cpu sorted | exclude 0.00%", "CPU占用"),
        ("show memory statistics", "內(nèi)存使用"),
        ("show logging | include %", "異常日志"),
        ("show ip route summary", "路由表概要"),
    ],
    "huawei": [
        ("display version | include VRP", "軟件版本"),
        ("display ip interface brief", "接口狀態(tài)"),
        ("display cpu-usage", "CPU占用"),
        ("display memory-usage", "內(nèi)存使用"),
        ("display logbuffer | include Error", "異常日志"),
        ("display ip routing-table statistics", "路由表概要"),
    ],
}

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

MAX_WORKERS = 10  # 同時(shí)連接 10 臺(tái)設(shè)備

def inspect_device(dev):
    """巡檢單臺(tái)設(shè)備，返回結(jié)果字典"""
    result = {
        "設(shè)備": dev["host"],
        "廠商": dev["device_type"],
        "巡檢時(shí)間": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "狀態(tài)": "成功",
    }

    # 合并憑據(jù)
    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"[執(zhí)行失敗] {e}"
    except Exception as e:
        result["狀態(tài)"] = "連接失敗"
        result["錯(cuò)誤信息"] = str(e)

    return result

# 并發(fā)執(zhí)行
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)} 臺(tái)設(shè)備")

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

wb = openpyxl.Workbook()
ws = wb.active
ws.title = f"設(shè)備巡檢報(bào)告_{datetime.now():%Y%m%d}"

# 表頭
headers = ["設(shè)備", "廠商", "狀態(tài)", "軟件版本", "接口狀態(tài)",
           "CPU占用", "內(nèi)存使用", "異常日志", "路由表概要"]
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")

# 異常標(biāo)記
red_fill = PatternFill(start_color="FFC7CE", end_color="FFC7CE",
                       fill_type="solid")
red_font = Font(color="9C0006")

# 寫入數(shù)據(jù)
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)

        # 包含異常關(guān)鍵詞則標(biāo)紅
        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

# 調(diào)整列寬
for col in ws.columns:
    ws.column_dimensions[col[0].column_letter].width = 22

filename = f"巡檢報(bào)告_{datetime.now():%Y%m%d}.xlsx"
wb.save(filename)

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("===== 設(shè)備版本信息 =====")
        print(conn.send_command("display version"))
except Exception as e:
    print("連接失敗：", e)

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("主機(jī)名:", facts["hostname"])
        print("型號(hào):", facts["model"])
        print("版本:", facts["os_version"])

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

        dev.discard_config()
except Exception as e:
    print("連接/執(zhí)行失敗:", str(e))

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"])

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)

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)

    # 華為等設(shè)備必須用交互式 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)}")