第302篇:Python 基础——网络工程师的第一行代码

关键词:Python基础、变量与数据类型、流程控制、函数、网络编程入门、正则表达式


一、引言

如果说网络自动化是一座大厦,Python就是它的钢筋混凝土。在全部网络自动化工具中——无论是Paramiko、Netmiko、NAPALM还是Scapy——Python都是最底层、最通用的语言基座。

对于习惯了CLI命令行的网络工程师而言,从零开始学习编程可能令人畏惧。但好消息是:你不需要成为软件工程师才能写出有用的网络脚本。本篇将从最基础的语法讲起,逐步构建编写网络自动化脚本所需的核心能力。


二、Python 环境准备与Hello World

2.1 安装与环境验证

# 推荐使用Python 3.9+,3.12为当前稳定版本
# Windows:从 python.org 下载安装包,勾选 "Add Python to PATH"
# Linux:sudo apt install python3 python3-pip

# 验证安装
python --version
# Python 3.12.4

# pip(包管理工具)验证
pip --version

2.2 第一个Python脚本

# hello_network.py
print("Hello, Network Automation!")

# 交互式运行
python hello_network.py
# 输出: Hello, Network Automation!

2.3 Jupyter Notebook:交互式编程利器

对于网络工程师,Jupyter Notebook提供"写一行、跑一行"的交互体验,非常适合探索性数据分析与设备数据可视化:

pip install jupyter
jupyter notebook

三、变量与数据类型

3.1 基本数据类型

# 整数(int) - 用于端口号、VLAN ID、IP地址段
vlan_id = 100
port_speed = 10000          # 10Gbps

# 浮点数(float) - 用于带宽利用率、丢包率
utilization = 78.56
packet_loss = 0.003

# 字符串(str) - 用于设备名、IP地址、命令
hostname = "Core-SW-01"
ip_address = "192.168.1.1"
command = "display interface brief"

# 布尔值(bool) - 用于链路状态、接口Up/Down
interface_up = True
bgp_established = False

3.2 列表(List):有序可变序列

列表是网络工程师最常用的数据结构,用于存储设备清单、命令列表等:

# 设备IP列表
switches = ["192.168.1.1", "192.168.1.2", "192.168.1.3"]

# 增
switches.append("192.168.1.4")

# 查
first_switch = switches[0]          # "192.168.1.1"
last_switch = switches[-1]          # "192.168.1.4"

# 删
switches.remove("192.168.1.2")

# 遍历
for ip in switches:
    print(f"连接到设备: {ip}")

3.3 字典(Dict):键值对映射

字典在网络自动化中无处不在——设备信息、配置参数、采集结果均以字典形式组织:

# 设备信息字典
device = {
    "hostname": "Core-SW-01",
    "vendor": "Huawei",
    "model": "CE6865",
    "management_ip": "192.168.1.1",
    "username": "admin",
    "password": "H@wa1e123",
    "role": "core"
}

# 按key取值
print(device["hostname"])       # Core-SW-01
print(device.get("serial"))     # None(key不存在时不报错)

# 修改
device["role"] = "spine"

# 遍历键值对
for key, value in device.items():
    print(f"{key}: {value}")

3.4 元组(Tuple)与集合(Set)

# 元组:不可变序列,适合存储固定组合
route_entry = ("10.0.0.0/8", "192.168.1.254", "GigabitEthernet0/0/1")
ip_port = ("192.168.1.1", 80)

# 集合:去重
all_ips = ["10.0.0.1", "10.0.0.2", "10.0.0.1", "10.0.0.3"]
unique_ips = set(all_ips)      # {"10.0.0.1", "10.0.0.2", "10.0.0.3"}

四、流程控制

4.1 条件判断

# 根据设备角色选择不同管理IP段
role = "core"

if role == "core":
    mgmt_subnet = "10.255.0.0/24"
elif role == "aggregation":
    mgmt_subnet = "10.255.1.0/24"
elif role == "access":
    mgmt_subnet = "10.255.2.0/24"
else:
    mgmt_subnet = "10.255.255.0/24"

print(f"管理网段: {mgmt_subnet}")

4.2 for 循环

# 批量设备连接演示
devices = [
    {"name": "SW-Core-01", "ip": "192.168.1.1"},
    {"name": "SW-Core-02", "ip": "192.168.1.2"},
    {"name": "SW-Access-01", "ip": "192.168.1.3"},
]

for device in devices:
    print(f"正在连接 {device['name']} ({device['ip']})...")
    # 此处将调用实际的SSH连接代码

4.3 while 循环

# 等待设备BGP会话Established
import time

max_retries = 10
retry = 0

while retry < max_retries:
    bgp_state = check_bgp_status(device_ip)   # 假设此函数返回当前状态

    if bgp_state == "Established":
        print("BGP会话已Established")
        break
    else:
        print(f"第{retry + 1}次检查: BGP状态 = {bgp_state},等待30秒...")
        time.sleep(30)
        retry += 1
else:
    print(f"警告: 超过{max_retries}次尝试,BGP仍未Established")

五、函数:封装可复用逻辑

5.1 函数定义与调用

# 定义函数
def ping_device(ip_address, count=4):
    """Ping指定IP地址,返回是否可达"""
    import subprocess
    result = subprocess.run(
        ["ping", "-c", str(count), ip_address],
        capture_output=True,
        text=True
    )
    return result.returncode == 0

# 调用函数
is_reachable = ping_device("192.168.1.1")
print(f"设备可达性: {is_reachable}")

5.2 处理设备信息的实用函数

import re

def parse_interface_line(line: str) -> dict:
    """解析 'display interface brief' 输出的一行"""
    # 示例行: GigabitEthernet0/0/1 * down  down   1000 baseX
    pattern = r'^(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)$'
    match = re.match(pattern, line.strip())

    if match:
        return {
            "interface": match.group(1),
            "physical": match.group(2),
            "protocol": match.group(3),
            "speed": match.group(4),
            "duplex": match.group(5),
            "type": match.group(6),
        }
    return None


def extract_vlans(config_text: str) -> list:
    """从配置文本中提取所有VLAN ID"""
    vlans = set()
    for line in config_text.splitlines():
        match = re.search(r'^\s*vlan\s+(\d+)', line)
        if match:
            vlans.add(int(match.group(1)))
    return sorted(vlans)


# 使用示例
config_snippet = """
vlan 10
 description USER_VLAN
vlan 20
 description SERVER_VLAN
vlan 100
 description MANAGEMENT
"""
print(extract_vlans(config_snippet))   # [10, 20, 100]

六、文件操作与网络数据采集

6.1 读取与写入文件

# 从文本文件读取设备清单
devices = []
with open("devices.txt", "r", encoding="utf-8") as f:
    for line in f:
        line = line.strip()
        if line and not line.startswith("#"):
            parts = line.split(",")
            devices.append({
                "hostname": parts[0],
                "ip": parts[1],
                "role": parts[2] if len(parts) > 2 else "unknown"
            })

# 将采集结果写入文件
with open("backup/switch-config.txt", "w", encoding="utf-8") as f:
    f.write(f"# 备份时间: 2025-01-15 10:30:00\n")
    f.write(f"# 设备: Core-SW-01\n")
    f.write(config_text)

6.2 处理CSV格式的设备清单

import csv

# 写入CSV
with open("device_inventory.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["hostname", "ip", "role", "vendor"])
    writer.writeheader()
    writer.writerows([
        {"hostname": "Core-SW-01", "ip": "192.168.1.1", "role": "core", "vendor": "Huawei"},
        {"hostname": "Core-SW-02", "ip": "192.168.1.2", "role": "core", "vendor": "Huawei"},
    ])

# 读取CSV
with open("device_inventory.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['hostname']}: {row['ip']} ({row['role']})")

七、正则表达式:网络数据解析利器

网络设备的CLI输出格式固定但不规范,正则表达式是最有效的解析工具:

import re

# 1. 提取IP地址
text = "The IP address for the interface is 10.1.1.1/24"
match = re.search(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})/(\d+)', text)
if match:
    print(f"IP: {match.group(1)}, Mask: /{match.group(2)}")

# 2. 解析BGP路由表条目
bgp_line = "* > 10.0.0.0/8    192.168.1.1     0    100    i"
pattern = r'\*\s*>\s+(\S+)\s+(\S+)\s+(\d+)\s+(\d+)\s+(\w)'
match = re.match(pattern, bgp_line)
if match:
    print(f"Prefix: {match.group(1)}")
    print(f"Nexthop: {match.group(2)}")
    print(f"Metric: {match.group(3)}")
    print(f"LocPrf: {match.group(4)}")
    print(f"Path: {match.group(5)}")

# 3. 匹配接口名称
interfaces = [
    "GigabitEthernet0/0/1",
    "XGigabitEthernet0/0/10",
    "Ten-GigabitEthernet1/0/0",
    "ethernet-0/0/1",
    "Gig0/1",
]
for iface in interfaces:
    match = re.match(r'(\w+)(\d+/\d+(?:/\d+)?)', iface)
    if match:
        print(f"类型: {match.group(1)}, 编号: {match.group(2)}")

八、异常处理与日志

import logging
import sys

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler("network_automation.log"),
        logging.StreamHandler(sys.stdout)
    ]
)

def connect_and_run(ip: str, command: str):
    """连接设备并执行命令,包含完整异常处理"""
    try:
        # 模拟SSH连接(实际使用Paramiko/Netmiko)
        if not ip:
            raise ValueError("IP地址不能为空")

        logging.info(f"正在连接设备: {ip}")

        # 模拟执行
        if ip == "192.168.1.99":
            raise ConnectionError(f"无法连接到 {ip}(超时)")

        result = f"命令 '{command}' 执行成功"
        logging.info(result)
        return result

    except ConnectionError as e:
        logging.error(f"连接异常: {e}")
        return None
    except ValueError as e:
        logging.error(f"参数异常: {e}")
        return None
    except Exception as e:
        logging.exception(f"未知异常: {e}")
        return None

# 测试
connect_and_run("192.168.1.1", "display ip routing-table")
connect_and_run("192.168.1.99", "display ip routing-table")

九、pip 包管理与虚拟环境

# 创建虚拟环境(隔离项目依赖)
python -m venv network-env

# 激活虚拟环境
# Windows:
network-env\Scripts\activate
# Linux/Mac:
source network-env/bin/activate

# 安装网络自动化常用包
pip install paramiko netmiko napalm textfsm jinja2 pyyaml

# 查看已安装包
pip list

# 导出依赖清单
pip freeze > requirements.txt

# 从清单恢复环境
pip install -r requirements.txt

# 退出虚拟环境
deactivate

十、总结

知识点 应用场景
变量与数据类型 存储设备IP、配置参数、采集结果
列表与字典 设备清单、配置模板变量、结构化数据
条件与循环 批量处理设备、按角色分类操作
函数 封装可复用逻辑(连接、采集、解析)
文件操作 读取设备清单、写入配置备份
正则表达式 解析CLI输出、提取关键信息
异常处理 连接超时、认证失败、命令错误
虚拟环境 管理项目依赖,避免包冲突

思考题

  1. 为什么字典(Dict)比列表(List)更适合存储设备信息?如果设备有50个属性,分别用列表和字典存储,哪种更容易维护?
  2. 编写一个函数,接收一个包含多行CLI输出的字符串,提取其中所有以GigabitEthernet开头的接口名称并返回列表。
  3. 正则表达式 r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' 能否匹配到999.999.999.999这样的无效IP?如何改进使其只匹配合法IP地址?

下篇预告

第303篇:Paramiko 库——SSH 连接网络设备。学习如何通过Python建立SSH会话,远程登录交换机和路由器执行命令与下发配置。