Python进阶——运维工程师的瑞士军刀

一、函数——把重复劳动封装成"一键操作"

1.1 问题引入

场景：你写了50行代码来检查服务器CPU使用率，现在需要在10台服务器上分别执行。复制粘贴50次？一旦逻辑要改，就得改50处。

解决思路：把检查逻辑封装成函数，一次定义、多处调用。

1.2 def定义函数与参数

# 基本函数定义
def check_disk_usage(mount_point, threshold=80):
    """检查磁盘使用率，超过阈值则告警"""
    import shutil
    usage = shutil.disk_usage(mount_point)
    percent = usage.used / usage.total * 100
    if percent > threshold:
        print(f"[告警] {mount_point} 使用率 {percent:.1f}%，超过阈值 {threshold}%")
        return False
    print(f"[正常] {mount_point} 使用率 {percent:.1f}%")
    return True

# 调用方式
check_disk_usage("/")              # 使用默认阈值80
check_disk_usage("/data", 90)      # 自定义阈值90
check_disk_usage(threshold=70, mount_point="/home")  # 关键字参数

参数类型一览：

1.3 *args 与 **kwargs

# 运维场景：批量检查多台服务器
def batch_check(action, *servers, **options):
    """
    action: 检查动作
    *servers: 任意数量的服务器IP
    **options: 额外选项（超时、端口等）
    """
    timeout = options.get("timeout", 5)
    port = options.get("port", 22)
    for server in servers:
        print(f"正在{action}服务器 {server}:{port}，超时{timeout}s")

# 调用
batch_check("ping", "192.168.1.1", "192.168.1.2", "192.168.1.3", timeout=3, port=22)

互动提问：如果调用 batch_check("ping", timeout=3)，servers 元组里有什么？

答案：空元组 ()，因为没有传入位置参数作为服务器地址。

1.4 return返回值

def parse_log_line(line):
    """解析日志行，返回结构化数据"""
    parts = line.strip().split()
    if len(parts) < 3:
        return None  # 无效日志行
    timestamp = parts[0]
    level = parts[1]
    message = " ".join(parts[2:])
    return {"time": timestamp, "level": level, "msg": message}  # 返回字典

result = parse_log_line("2024-01-15 ERROR 数据库连接超时")
if result:
    print(f"时间: {result['time']}, 级别: {result['level']}")

1.5 lambda匿名函数

# 场景：按文件大小排序
import os
files = os.listdir("/var/log")
# lambda 适合简单的、一次性的逻辑
files_sorted = sorted(files, key=lambda f: os.path.getsize(f"/var/log/{f}"), reverse=True)
print("最大的5个日志文件:", files_sorted[:5])

类比：lambda 就像便利贴——写个简短的备忘，用完就丢，不需要正式定义一个函数。

1.6 变量作用域与global/nonlocal

# LEGB规则：Local → Enclosing → Global → Built-in
server_count = 0  # 全局变量

def monitor_team():
    team_servers = 10  # 外层函数变量

    def add_server():
        nonlocal team_servers  # 声明使用外层变量
        global server_count     # 声明使用全局变量
        team_servers += 1
        server_count += 1
        print(f"团队服务器: {team_servers}, 总服务器: {server_count}")

    add_server()
    add_server()

monitor_team()

互动提问：如果不写 nonlocal team_servers，直接 team_servers += 1 会怎样？

答案：报 UnboundLocalError，Python 会认为 team_servers 是局部变量但还没赋值就被使用了。

1.7 递归函数

# 运维场景：递归遍历目录树，查找特定文件
def find_config_files(directory, target_ext=".conf", depth=0):
    """递归查找配置文件"""
    import os
    indent = "  " * depth
    try:
        for item in os.listdir(directory):
            full_path = os.path.join(directory, item)
            if os.path.isfile(full_path) and item.endswith(target_ext):
                print(f"{indent}[找到] {full_path}")
            elif os.path.isdir(full_path):
                find_config_files(full_path, target_ext, depth + 1)
    except PermissionError:
        print(f"{indent}[权限不足] {directory}")

find_config_files("/etc", ".conf")

运维实战案例：

# 综合案例：服务器健康检查函数
import subprocess

def health_check(host, checks=("cpu", "mem", "disk")):
    """综合健康检查"""
    results = {}
    check_cmds = {
        "cpu": "top -bn1 | grep 'Cpu(s)' | awk '{print $2}'",
        "mem": "free -m | awk 'NR==2{printf \"%.1f\", $3*100/$2}'",
        "disk": "df -h / | awk 'NR==2{print $5}'"
    }
    for check in checks:
        try:
            cmd = check_cmds.get(check, "echo unknown")
            output = subprocess.check_output(cmd, shell=True, timeout=10)
            results[check] = output.decode().strip()
        except Exception as e:
            results[check] = f"错误: {e}"
    return results

print(health_check("localhost"))

二、模块——站在巨人的肩膀上

2.1 问题引入

场景：你想获取当前系统进程列表、解析命令行参数、复制文件夹……每个都从零写？太慢了！

解决思路：Python有丰富的内置模块和第三方模块，学会 import 就是学会"借力"。

2.2 import的多种方式

# 方式1：导入整个模块
import os
print(os.getcwd())

# 方式2：导入特定函数
from os.path import join, exists
path = join("/var", "log", "syslog")
print(exists(path))

# 方式3：起别名（常用于避免命名冲突）
import subprocess as sp
result = sp.run(["uptime"], capture_output=True, text=True)

# 方式4：导入所有（不推荐，容易命名冲突）
# from os import *

2.3 name 与自定义模块

# 文件：server_utils.py
"""服务器工具模块"""

def get_uptime():
    """获取系统运行时间"""
    with open("/proc/uptime") as f:
        uptime_seconds = float(f.readline().split()[0])
    hours = uptime_seconds / 3600
    return f"系统已运行 {hours:.1f} 小时"

def get_load_average():
    """获取系统负载"""
    import os
    load1, load5, load15 = os.getloadavg()
    return {"1min": load1, "5min": load5, "15min": load15}

# 只有直接运行时才执行，被import时不执行
if __name__ == "__main__":
    print(get_uptime())
    print(get_load_average())

类比：__name__ == "__main__" 就像餐厅厨房的"试菜"——厨师自己做着尝（直接运行）会执行试菜流程，但客人点菜时（被import）就直接上菜不走试菜。

2.4 运维常用模块详解

os模块——操作系统接口

import os

# 文件与目录操作
os.getcwd()                        # 当前工作目录
os.listdir("/etc")                 # 列出目录内容
os.makedirs("/tmp/test/a/b", exist_ok=True)  # 递归创建目录
os.rename("/tmp/old.log", "/tmp/new.log")    # 重命名
os.environ.get("HOME")             # 获取环境变量
os.getpid()                        # 当前进程ID

sys模块——解释器相关

import sys

sys.argv           # 命令行参数列表
sys.version        # Python版本信息
sys.platform       # 平台标识（linux/win32/darwin）
sys.exit(1)        # 退出程序，1表示异常退出
sys.path           # 模块搜索路径

subprocess模块——执行系统命令

import subprocess

# 推荐方式：subprocess.run()（Python 3.5+）
result = subprocess.run(
    ["df", "-h"],
    capture_output=True,  # 捕获输出
    text=True,            # 返回字符串而非bytes
    timeout=30            # 超时时间
)
print(f"返回码: {result.returncode}")
print(f"输出:\n{result.stdout}")

# 管道操作：相当于 ps aux | grep nginx
ps = subprocess.Popen(["ps", "aux"], stdout=subprocess.PIPE)
grep = subprocess.Popen(["grep", "nginx"], stdin=ps.stdout, stdout=subprocess.PIPE, text=True)
output = grep.communicate()[0]

shutil模块——高级文件操作

import shutil

shutil.copy2("/etc/hosts", "/tmp/hosts.bak")     # 复制（保留元数据）
shutil.copytree("/etc/nginx", "/tmp/nginx.bak")  # 递归复制整个目录
shutil.move("/tmp/old.log", "/archive/old.log")  # 移动文件
shutil.rmtree("/tmp/test")                        # 递归删除目录
shutil.which("nginx")                             # 查找命令的完整路径

互动提问：os.remove() 和 shutil.rmtree() 有什么区别？

答案：os.remove() 只能删除单个文件；shutil.rmtree() 能递归删除整个目录树（包括子目录和文件）。

三、异常处理——让脚本"摔不坏"

3.1 问题引入

场景：你写了一个监控脚本，每5分钟检查一次网站是否在线。某天网络抖动，脚本直接崩溃，后面100台服务器的检查全部跳过。

解决思路：用异常处理"兜底"，即使单台出错也继续检查其他服务器。

3.2 try/except/else/finally

import subprocess

def check_service(service_name):
    """检查服务状态，完善异常处理"""
    try:
        result = subprocess.run(
            ["systemctl", "is-active", service_name],
            capture_output=True, text=True, timeout=5
        )
        # 如果命令返回非0，抛出异常
        if result.returncode != 0:
            raise RuntimeError(f"{service_name} 不是active状态")
    except FileNotFoundError:
        print(f"[错误] systemctl命令不存在，可能不是Linux系统")
    except subprocess.TimeoutExpired:
        print(f"[超时] 检查 {service_name} 超时5秒")
    except RuntimeError as e:
        print(f"[告警] {e}")
    except Exception as e:
        print(f"[未知错误] {type(e).__name__}: {e}")
    else:
        # try块没有异常时执行
        print(f"[正常] {service_name} 运行中")
    finally:
        # 无论是否异常都会执行
        print(f"[记录] 已完成 {service_name} 的检查")

# 批量检查，单台出错不影响其他
services = ["nginx", "mysql", "redis", "docker"]
for svc in services:
    check_service(svc)
    print("---")

3.3 raise主动抛出与自定义异常

# 自定义异常类
class DiskSpaceError(Exception):
    """磁盘空间异常"""
    def __init__(self, mount_point, usage_percent):
        self.mount_point = mount_point
        self.usage = usage_percent
        super().__init__(f"磁盘 {mount_point} 使用率 {usage_percent}% 超标！")

def check_disk(mount_point, threshold=90):
    """检查磁盘并主动抛出异常"""
    import shutil
    usage = shutil.disk_usage(mount_point)
    percent = int(usage.used / usage.total * 100)
    if percent > threshold:
        raise DiskSpaceError(mount_point, percent)
    return percent

# 使用自定义异常
try:
    check_disk("/", threshold=50)
except DiskSpaceError as e:
    print(f"运维告警: {e}")
    print(f"挂载点: {e.mount_point}, 使用率: {e.usage}%")

互动提问：except Exception as e 能捕获 KeyboardInterrupt（Ctrl+C）吗？

答案：不能！KeyboardInterrupt 和 SystemExit 继承自 BaseException 而不是 Exception，所以不会被 except Exception 捕获。这是Python故意这样设计的，确保你可以随时用Ctrl+C中断程序。

3.4 运维实战：日志轮转异常安全脚本

import os
import shutil
from datetime import datetime

def safe_log_rotate(log_dir, max_days=30):
    """安全的日志轮转，异常不会中断整个过程"""
    rotated = 0
    errors = 0
    for filename in os.listdir(log_dir):
        filepath = os.path.join(log_dir, filename)
        try:
            if not os.path.isfile(filepath):
                continue
            # 检查文件修改时间
            mtime = os.path.getmtime(filepath)
            age_days = (datetime.now().timestamp() - mtime) / 86400
            if age_days > max_days:
                archive_dir = os.path.join(log_dir, "archive")
                os.makedirs(archive_dir, exist_ok=True)
                shutil.move(filepath, os.path.join(archive_dir, filename))
                rotated += 1
        except PermissionError:
            print(f"[权限不足] 跳过: {filename}")
            errors += 1
        except Exception as e:
            print(f"[出错] {filename}: {e}")
            errors += 1
    print(f"轮转完成: 归档 {rotated} 个, 失败 {errors} 个")

四、正则表达式——文本处理的"瑞士军刀"

4.1 问题引入

场景：你有10万行Nginx访问日志，需要找出所有返回500错误且来自某个IP段的请求。用字符串的 find/split？写出来又长又脆弱。

解决思路：正则表达式用模式匹配文本，一行代码搞定复杂提取。

4.2 re模块核心方法

import re

log_line = '192.168.1.100 - - [15/Jan/2024:10:30:45 +0800] "GET /api/users HTTP/1.1" 500 1234'

# match：从字符串开头匹配
m = re.match(r'(\d+\.\d+\.\d+\.\d+)', log_line)
if m:
    print(f"IP地址: {m.group(1)}")  # 192.168.1.100

# search：在字符串中搜索（不限开头）
m = re.search(r'" (\d{3}) ', log_line)
if m:
    print(f"HTTP状态码: {m.group(1)}")  # 500

# findall：找到所有匹配
ips = re.findall(r'\d+\.\d+\.\d+\.\d+', log_line)
print(f"所有IP: {ips}")

# sub：替换
cleaned = re.sub(r'\d+\.\d+\.\d+\.\d+', '[IP已隐藏]', log_line)
print(f"脱敏: {cleaned}")

4.3 编译模式（提升性能）

import re

# 编译正则对象——在循环中重复使用时性能更好
ip_pattern = re.compile(r'(\d{1,3}\.){3}\d{1,3}')
status_pattern = re.compile(r'" (\d{3}) ')
error_pattern = re.compile(r'" [45]\d{2} ')

# 运维场景：分析访问日志
def analyze_access_log(log_file):
    error_count = 0
    ip_errors = {}

    with open(log_file, 'r') as f:
        for line in f:
            ip_match = ip_pattern.search(line)
            if error_pattern.search(line) and ip_match:
                ip = ip_match.group()
                error_count += 1
                ip_errors[ip] = ip_errors.get(ip, 0) + 1

    print(f"总错误请求数: {error_count}")
    print("错误最多的IP:")
    for ip, count in sorted(ip_errors.items(), key=lambda x: -x[1])[:5]:
        print(f"  {ip}: {count}次")

类比：正则表达式就像"通缉令"上的特征描述——"身高170-180cm、短发、戴眼镜"，re模块拿着这个描述去人群中搜索匹配的人。

4.4 运维常用正则

# 匹配IP地址（精确）
ip_re = re.compile(r'\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b')

# 匹配日期时间
datetime_re = re.compile(r'\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}')

# 匹配邮箱
email_re = re.compile(r'[\w.+-]+@[\w-]+\.[\w.]+')

# 提取配置文件中的键值对（如 nginx.conf）
kv_re = re.compile(r'^\s*(\w+)\s+(.+?);\s*$')

互动提问：re.match 和 re.search 的区别是什么？

答案：match 只从字符串开头匹配；search 在整个字符串中搜索第一个匹配。例如 re.match("abc", "xyzabc") 返回 None，而 re.search("abc", "xyzabc") 能匹配到。

五、文件IO——与操作系统对话的桥梁

5.1 问题引入

场景：你需要读取配置文件、写入日志、遍历目录查找大文件……这些都是文件IO操作。

5.2 open与with语句

# 传统方式（容易忘记关闭文件）
f = open("/etc/hostname", "r")
content = f.read()
f.close()  # 如果中间出异常，这行不会执行！

# 推荐方式：with语句（自动关闭文件）
with open("/etc/hostname", "r") as f:
    content = f.read().strip()
    print(f"主机名: {content}")
# 离开with块后文件自动关闭，即使发生异常也会关闭

读写模式速查表：

5.3 读写操作实战

# 写日志
from datetime import datetime

def write_log(log_file, level, message):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = f"[{timestamp}] [{level}] {message}\n"
    with open(log_file, "a", encoding="utf-8") as f:
        f.write(log_entry)

write_log("/tmp/app.log", "INFO", "服务启动成功")
write_log("/tmp/app.log", "ERROR", "数据库连接失败")

# 读取配置文件
def read_config(config_file):
    config = {}
    with open(config_file, "r") as f:
        for line in f:
            line = line.strip()
            if line and not line.startswith("#"):
                key, value = line.split("=", 1)
                config[key.strip()] = value.strip()
    return config

5.4 os.path与os.walk

import os

# os.path 常用操作
os.path.join("/var", "log", "nginx", "access.log")  # 路径拼接
os.path.exists("/etc/nginx/nginx.conf")              # 路径是否存在
os.path.isfile("/etc/hosts")                          # 是否是文件
os.path.isdir("/var/log")                             # 是否是目录
os.path.basename("/var/log/nginx/access.log")         # 文件名：access.log
os.path.dirname("/var/log/nginx/access.log")          # 目录名：/var/log/nginx
os.path.getsize("/var/log/syslog")                    # 文件大小（字节）
os.path.splitext("backup.tar.gz")                     # 分离扩展名

# os.walk 遍历目录树
def find_large_files(root_dir, size_mb=100):
    """查找大于指定大小的文件"""
    large_files = []
    for dirpath, dirnames, filenames in os.walk(root_dir):
        for filename in filenames:
            filepath = os.path.join(dirpath, filename)
            try:
                size = os.path.getsize(filepath)
                if size > size_mb * 1024 * 1024:
                    large_files.append((filepath, size / (1024*1024)))
            except (PermissionError, OSError):
                continue
    large_files.sort(key=lambda x: x[1], reverse=True)
    return large_files

for path, size in find_large_files("/var/log", 50):
    print(f"{size:.1f} MB  {path}")

互动提问：os.walk() 返回的三个值分别是什么？

答案：dirpath（当前目录路径）、dirnames（子目录名列表）、filenames（文件名列表）。它会自动递归进入所有子目录。

六、反射——运行时动态操作对象

6.1 问题引入

场景：你开发了一个运维平台，用户在Web界面输入命令名（如"check_cpu"），后端需要根据字符串动态调用对应的函数。

解决思路：用反射机制，根据字符串获取对象的属性或方法。

6.2 getattr/hasattr/setattr

class ServerManager:
    def __init__(self, host):
        self.host = host
        self.status = "unknown"

    def check_cpu(self):
        return f"{self.host} CPU使用率: 45%"

    def check_memory(self):
        return f"{self.host} 内存使用率: 62%"

    def check_disk(self):
        return f"{self.host} 磁盘使用率: 78%"

sm = ServerManager("web-server-01")

# hasattr：检查属性/方法是否存在
print(hasattr(sm, "check_cpu"))   # True
print(hasattr(sm, "check_gpu"))   # False

# getattr：通过字符串获取属性/方法并调用
command = "check_cpu"  # 来自用户输入
if hasattr(sm, command):
    method = getattr(sm, command)
    print(method())  # 输出: web-server-01 CPU使用率: 45%
else:
    print(f"不支持的命令: {command}")

# setattr：动态设置属性
setattr(sm, "region", "北京机房")
print(sm.region)  # 北京机房

类比：反射就像自动售货机——你按按钮（字符串），机器找到对应的格子（属性/方法）然后出货。你不需要知道每个格子的内部结构，只需要知道按钮名字。

七、装饰器——给函数穿上"增强装备"

7.1 问题引入

场景：你有20个运维函数，每个都需要：记录日志、计算执行时间、检查权限。每个函数里都写一遍？太啰嗦。

解决思路：装饰器把公共逻辑抽出来，一行 @装饰器 就能增强任意函数。

7.2 闭包——装饰器的基础

# 闭包：内层函数引用外层函数的变量
def make_multiplier(factor):
    def multiplier(x):
        return x * factor  # 引用了外层的factor
    return multiplier

double = make_multiplier(2)
triple = make_multiplier(3)
print(double(5))   # 10
print(triple(5))   # 15

7.3 基础装饰器

import time
import functools

def timer(func):
    """计时装饰器：统计函数执行时间"""
    @functools.wraps(func)  # 保留原函数的名字和文档
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        elapsed = time.time() - start
        print(f"[计时] {func.__name__} 执行耗时: {elapsed:.3f}秒")
        return result
    return wrapper

def logger(func):
    """日志装饰器：记录函数调用"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f"[日志] 调用 {func.__name__}，参数: args={args}, kwargs={kwargs}")
        result = func(*args, **kwargs)
        print(f"[日志] {func.__name__} 执行完毕")
        return result
    return wrapper

# 使用装饰器（可以叠加多个）
@timer
@logger
def backup_database(db_name):
    """备份数据库"""
    time.sleep(1)  # 模拟耗时操作
    print(f"正在备份数据库: {db_name}")
    return f"{db_name}_backup.sql"

result = backup_database("production_db")

7.4 带参数的装饰器

import functools

def retry(max_attempts=3, delay=2):
    """重试装饰器（带参数）"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"[重试] {func.__name__} 第{attempt}次失败: {e}")
                    if attempt < max_attempts:
                        print(f"[重试] {delay}秒后重试...")
                        time.sleep(delay)
                    else:
                        print(f"[失败] {func.__name__} 已达最大重试次数")
                        raise
        return wrapper
    return decorator

import time

@retry(max_attempts=5, delay=1)
def check_website(url):
    """检查网站是否可达"""
    import urllib.request
    response = urllib.request.urlopen(url, timeout=5)
    return response.status

# 调用
try:
    status = check_website("http://example.com")
    print(f"状态码: {status}")
except Exception:
    print("网站最终不可达")

互动提问：装饰器的本质是什么？

答案：装饰器就是一个接受函数作为参数、返回新函数的函数。@timer 等价于 backup_database = timer(backup_database)。

7.5 运维实战：权限检查装饰器

import os
import functools

def require_root(func):
    """要求root权限才能执行"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if os.geteuid() != 0:
            raise PermissionError(f"[权限不足] {func.__name__} 需要root权限执行！")
        return func(*args, **kwargs)
    return wrapper

@require_root
def restart_service(service_name):
    import subprocess
    subprocess.run(["systemctl", "restart", service_name], check=True)
    print(f"{service_name} 已重启")

八、面向对象——用"类"组织复杂系统

8.1 问题引入

场景：你要管理100台服务器，每台有IP、角色、状态等信息，还有重启、备份、监控等操作。用字典+散落函数？越写越乱。

解决思路：面向对象把数据（属性）和操作（方法）打包成"类"，就像蓝图与建筑的关系。

8.2 class定义与__init__

class Server:
    """服务器类"""

    # 类属性（所有实例共享）
    datacenter = "北京亦庄机房"
    total_count = 0

    def __init__(self, hostname, ip, role="web"):
        """构造方法：创建实例时自动调用"""
        self.hostname = hostname    # 实例属性
        self.ip = ip
        self.role = role
        self.status = "running"
        Server.total_count += 1

    def __str__(self):
        """定义print()时的输出"""
        return f"Server({self.hostname}, {self.ip}, {self.role}, {self.status})"

    def restart(self):
        print(f"正在重启 {self.hostname}...")
        self.status = "restarting"
        # 模拟重启过程
        self.status = "running"
        print(f"{self.hostname} 重启完成")

    def get_info(self):
        return {
            "hostname": self.hostname,
            "ip": self.ip,
            "role": self.role,
            "status": self.status,
            "datacenter": self.datacenter
        }

# 创建实例
web1 = Server("web-01", "192.168.1.10", "web")
db1 = Server("db-01", "192.168.1.20", "database")
print(web1)                    # Server(web-01, 192.168.1.10, web, running)
print(f"服务器总数: {Server.total_count}")  # 2

8.3 继承与多态

class WebServer(Server):
    """Web服务器——继承自Server"""

    def __init__(self, hostname, ip, web_server="nginx"):
        super().__init__(hostname, ip, role="web")  # 调用父类构造
        self.web_server = web_server

    def deploy(self, app_name):
        print(f"在 {self.hostname} 上部署 {app_name}（{self.web_server}）")

    def restart(self):
        """重写父类方法"""
        print(f"优雅重启 {self.hostname} 的 {self.web_server}...")
        self.status = "restarting"
        self.status = "running"

class DatabaseServer(Server):
    """数据库服务器"""

    def __init__(self, hostname, ip, db_type="mysql"):
        super().__init__(hostname, ip, role="database")
        self.db_type = db_type

    def backup(self):
        print(f"备份 {self.hostname} 的 {self.db_type} 数据...")

    def restart(self):
        print(f"安全关闭 {self.hostname} 的 {self.db_type} 后重启...")

# 多态：不同类型的服务器，调用同一个方法，行为不同
servers = [
    WebServer("web-01", "10.0.0.1", "nginx"),
    DatabaseServer("db-01", "10.0.0.2", "mysql"),
    WebServer("web-02", "10.0.0.3", "apache"),
]

for server in servers:
    server.restart()  # 每个服务器类型有自己的重启方式

8.4 封装——保护内部数据

class ConfigManager:
    """配置管理器（演示封装）"""

    def __init__(self):
        self._config = {}       # 单下划线：约定为内部使用
        self.__secret = ""      # 双下划线：名称改写，外部不能直接访问

    def load_config(self, filepath):
        """加载配置"""
        with open(filepath) as f:
            for line in f:
                if "=" in line and not line.startswith("#"):
                    k, v = line.strip().split("=", 1)
                    self._config[k.strip()] = v.strip()

    def get(self, key, default=None):
        return self._config.get(key, default)

    def set_password(self, password):
        """设置密码（通过方法控制访问）"""
        if len(password) < 8:
            raise ValueError("密码长度至少8位")
        self.__secret = password

互动提问：Python的"封装"和Java的 private 一样严格吗？

答案：不一样。Python的封装是"君子协定"而非强制限制。双下划线 __name 只是做了名称改写（name mangling），仍然可以通过 _ClassName__name 访问。Python信奉"我们都是成年人"，靠约定而非强制。

8.5 运维实战：服务器集群管理类

class Cluster:
    """服务器集群管理"""

    def __init__(self, name):
        self.name = name
        self.servers = []

    def add_server(self, server):
        if not isinstance(server, Server):
            raise TypeError("只能添加Server或其子类的实例")
        self.servers.append(server)
        print(f"[{self.name}] 添加服务器: {server.hostname}")

    def get_by_role(self, role):
        return [s for s in self.servers if s.role == role]

    def status_report(self):
        print(f"\n{'='*50}")
        print(f"集群: {self.name}")
        print(f"{'='*50}")
        print(f"服务器总数: {len(self.servers)}")
        for server in self.servers:
            print(f"  {server}")
        print(f"{'='*50}\n")

# 使用
cluster = Cluster("生产集群-A")
cluster.add_server(WebServer("web-01", "10.0.0.1"))
cluster.add_server(WebServer("web-02", "10.0.0.2"))
cluster.add_server(DatabaseServer("db-01", "10.0.0.3"))
cluster.status_report()
print(f"Web服务器: {len(cluster.get_by_role('web'))} 台")

运维常用命令精选

Python运维脚本常用片段

# 一行命令启动HTTP文件服务器（分享文件）
python3 -m http.server 8080

# JSON格式化处理配置文件
python3 -m json.tool < config.json

# 快速计算文件MD5
python3 -c "import hashlib; print(hashlib.md5(open('file.txt','rb').read()).hexdigest())"

# 批量重命名文件
python3 -c "
import os, glob
for f in glob.glob('*.log'):
    os.rename(f, f.replace('.log', '.log.bak'))
"

# 端口扫描
python3 -c "
import socket
for port in [22, 80, 443, 3306, 6379, 8080]:
    s = socket.socket()
    s.settimeout(1)
    result = s.connect_ex(('192.168.1.1', port))
    status = '开放' if result == 0 else '关闭'
    print(f'端口 {port}: {status}')
    s.close()
"

pip包管理常用命令

pip install requests              # 安装第三方库
pip install -r requirements.txt   # 从文件安装依赖
pip list                          # 已安装的包
pip show requests                 # 查看包详情
pip freeze > requirements.txt     # 导出依赖列表
pip install --upgrade pip         # 升级pip

趣味命令

# Python之禅——Python的设计哲学
python3 -c "import this"

# 用Python画一只乌龟（Turtle图形）
python3 -c "
import turtle
t = turtle.Turtle()
for i in range(4):
    t.forward(100)
    t.right(90)
turtle.done()
"

# 用Python生成二维码（需安装qrcode库）
# pip install qrcode[pil]
python3 -c "
import qrcode
img = qrcode.make('Hello, 运维世界!')
img.save('hello_ops.png')
print('二维码已保存为 hello_ops.png')
"

# Python版系统监控仪表盘（一行版）
python3 -c "
import os, shutil, platform
print(f'系统: {platform.system()} {platform.release()}')
print(f'CPU核心: {os.cpu_count()}')
mem_info = os.popen('free -h').read() if os.name != 'nt' else 'N/A'
print(f'内存:\n{mem_info}')
d = shutil.disk_usage('/')
print(f'磁盘(/): 总{d.total//2**30}G, 已用{d.used//2**30}G, 可用{d.free//2**30}G')
"