Python Linux 系统运维详解

核心概念 Python 适合复杂、可维护、批量化的自动化运维脚本，覆盖服务器信息采集、进程管理、网络监控、日志分析等。

1. Python 运维核心定位

Shell 适合单行简单命令；Python 适合复杂、可维护、批量、有逻辑、带日志、异常处理、结构化输出的自动化运维脚本。

核心覆盖：

服务器信息采集、文件/目录管理、进程管理、磁盘/内存/网络监控、定时任务、批量远程主机操作、日志分析、服务启停、系统参数调优、容器基础操作。

运维必备四大标准库：

1. os：底层系统、目录、文件、进程基础操作
2. shutil：高级文件复制、移动、递归删除（os 增强）
3. subprocess：执行 Linux shell 命令，采集命令输出（运维最常用）
4. psutil：跨平台系统资源采集（CPU/内存/磁盘/网络/进程，运维神器）
   扩展高频库：

• paramiko：SSH 远程批量操作多台 Linux
• configparser/tomli：读取配置文件
• logging：脚本日志落地，线上必备
• datetime/time：时间戳、定时判断、告警时间
• re：日志正则匹配、过滤报错、提取指标
• json：结构化指标输出，对接监控平台

二、四大核心运维标准库深度讲解

2. subprocess：执行 Linux 命令（运维第一核心）

所有查看系统状态、启停服务、df/top/netstat/curl 全部依靠它。

2.1. 核心规则

1. shell=False：列表传参，无注入风险，推荐；
2. shell=True：支持管道 |、通配符 *，不可接收外部用户输入；
3. capture_output=True 捕获 stdout/stderr；text=True 输出字符串，避免 bytes 解码；
4. timeout 防止命令卡死阻塞脚本；
5. check=True 命令失败直接抛异常，不用手动判断返回码。

2.2. 通用封装运维执行函数（生产直接复用）

import subprocess

def run_cmd(cmd_list, timeout=10):

    """执行命令，返回(返回码,标准输出,错误输出)"""

    try:

        res = subprocess.run(

            cmd_list,

            capture_output=True,

            text=True,

            timeout=timeout

        )

        return res.returncode, res.stdout.strip(), res.stderr.strip()

    except subprocess.TimeoutExpired as e:

        return -1, "", f"命令超时:{e}"

# 示例：查看磁盘

code, out, err = run_cmd(["df", "-h"])

if code == 0:

    print(out)

else:

    print("执行失败", err)

2.3. 管道场景（shell=True）

# 统计TCP连接数

code, out, err = run_cmd("ss -t | grep ESTAB | wc -l", shell=True)

print("TCP连接数", out)

2.4. 长输出实时打印（监控日志、tail -f）

def run_stream_cmd(cmd):

    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True)

    while line := p.stdout.readline():

        print(line.strip())

run_stream_cmd(["tail", "-f", "/var/log/messages"])

3. psutil：系统资源监控采集（运维监控核心）

无需调用 shell 命令，直接读取内核数据，采集性能指标，写监控告警脚本首选。

3.1. 安装

pip3 install psutil

3.2. 1）CPU 信息

import psutil

# 总使用率

psutil.cpu_percent(interval=1)

# 核心数量

psutil.cpu_count()

# 每个核心使用率

psutil.cpu_percent(percpu=True)

3.3. 2）内存

mem = psutil.virtual_memory()

print(f"总内存{mem.total/1024/1024:.0f}M")

print(f"可用{mem.available/1024/1024:.0f}M")

print(f"使用率{mem.percent}%")

# 交换分区

swap = psutil.swap_memory()

3.4. 3）磁盘分区、磁盘IO

# 所有分区

partitions = psutil.disk_partitions()

for p in partitions:

    mount = p.mountpoint

    usage = psutil.disk_usage(mount)

    print(mount, f"使用率{usage.percent}%")

# 磁盘IO读写

io = psutil.disk_io_counters()

io.read_bytes, io.write_bytes

3.5. 4）网络、连接、网卡流量

# 网卡总收发

net = psutil.net_io_counters()

# 所有TCP连接

conns = psutil.net_connections(kind="tcp")

3.6. 5）进程管理（查占用CPU/内存最高进程、杀进程）

# 遍历所有进程

for proc in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):

    info = proc.info

    print(info["pid"], info["name"], info["cpu_percent"])

# 根据PID杀死进程

p = psutil.Process(1234)

p.terminate()

# 强制杀

# p.kill()

4. os + shutil：文件、目录、权限、系统基础操作

4.1. os 核心运维功能

import os

# 当前工作目录

os.getcwd()

# 切换目录

os.chdir("/data/log")

# 列出目录

os.listdir("/tmp")

# 判断文件/目录是否存在

os.path.exists("/etc/hosts")

os.path.isfile("/etc/hosts")

os.path.isdir("/data")

# 创建单层目录

os.mkdir("/tmp/test")

# 递归多层目录

os.makedirs("/tmp/a/b/c", exist_ok=True)

# 删除文件

os.remove("/tmp/1.txt")

# 删除空文件夹

os.rmdir("/tmp/test")

# 路径拼接（跨系统兼容）

file_path = os.path.join("/data", "app.log")

# 获取文件大小

os.path.getsize(file_path)

# 执行系统命令（废弃，优先subprocess）

os.system("ls -l")

# 获取环境变量

os.environ.get("PATH")

# 获取当前用户uid/gid

os.getuid()

4.2. shutil：高级文件操作（递归复制、移动、删除、打包）

import shutil

# 复制文件

shutil.copy("/etc/hosts", "/tmp/hosts.bak")

# 递归复制目录

shutil.copytree("/data/app", "/data/app_bak")

# 移动/重命名

shutil.move("/tmp/1.txt", "/data/1.txt")

# 递归删除目录（非空文件夹直接删）

shutil.rmtree("/tmp/test_dir")

# 压缩打包

shutil.make_archive("bak", "tar", "/data/app")

5. 文件读写、日志处理（with 迭代器逐行读取）

运维大量场景：分析日志、过滤报错、统计关键字、清理过期日志。

5.1. 超大日志逐行读取（不占内存）

def filter_error_log(log_path):

    error_lines = []

    with open(log_path, "r", encoding="utf-8", errors="replace") as f:

        for line in f:

            if "ERROR" in line or "Exception" in line:

                error_lines.append(line.strip())

    return error_lines

5.2. 日志清理：删除7天前日志

from pathlib import Path

import time

def clean_old_logs(log_dir, day=7):

    now = time.time()

    for file in Path(log_dir).iterdir():

        if not file.is_file():

            continue

        mtime = file.stat().st_mtime

        if now - mtime > day * 86400:

            file.unlink()

6. logging：运维脚本标准日志（线上必备）

shell 脚本只能简单echo，Python 日志支持分级、落地文件、按大小切割、时间切割、自动轮转。

import logging

from logging.handlers import RotatingFileHandler

def init_logger():

    logger = logging.getLogger("ops_script")

    logger.setLevel(logging.INFO)

    # 按文件大小切割，单个日志最大100M，保留5个备份

    handler = RotatingFileHandler(

        "/var/log/ops.log", maxBytes=100*1024*1024, backupCount=5, encoding="utf-8"

    )

    fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")

    handler.setFormatter(fmt)

    logger.addHandler(handler)

    return logger

logger = init_logger()

logger.info("脚本启动")

logger.error("磁盘使用率超过90%！")

三、远程批量运维核心：paramiko 多主机批量执行

场景：一次性操作几十台服务器批量重启服务、下发文件、采集指标。

7. 安装

pip3 install paramiko

8. 基础单台主机执行命令

import paramiko

def ssh_exec(host, user, pwd, cmd, port=22):

    ssh = paramiko.SSHClient()

    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:

        ssh.connect(host, username=user, password=pwd, port=port, timeout=5)

        stdin, stdout, stderr = ssh.exec_command(cmd)

        out = stdout.read().decode("utf-8", errors="replace")

        err = stderr.read().decode("utf-8", errors="replace")

        return True, out, err

    except Exception as e:

        return False, "", str(e)

    finally:

        ssh.close()

# 调用

ok, out, err = ssh_exec("192.168.1.100", "root", "123456", "df -h")

print(out)

9. 批量多主机循环执行

host_list = [

    {"host":"192.168.1.100","user":"root","pwd":"xxx"},

    {"host":"192.168.1.101","user":"root","pwd":"xxx"},

]

for host_info in host_list:

    ok, out, err = ssh_exec(host_info["host"], host_info["user"], host_info["pwd"], "free -h")

    print(f"====={host_info['host']}=====")

    print(out)

10. 上传/下载文件 SFTPClient

def ssh_upload(host, user, pwd, local, remote):

    ssh = paramiko.SSHClient()

    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    ssh.connect(host, username=user, password=pwd)

    sftp = ssh.open_sftp()

    sftp.put(local, remote) # 本地上传到服务器

    # sftp.get(remote, local) 服务器下载到本地

    sftp.close()

    ssh.close()

四、运维高频配套工具库

11. re 正则：日志提取指标、过滤异常

import re

# 提取日志中所有ERROR报错行

pattern = re.compile(r".*ERROR.*")

line = "2026-06-26 12:00:00 ERROR connect failed"

res = pattern.match(line)

if res:

    print("发现错误日志")

12. json：结构化输出指标，对接Prometheus/监控平台

import json

data = {

    "host":"server01",

    "cpu_used": 25,

    "mem_used": 60,

    "disk_warn": False

}

print(json.dumps(data, indent=2))

13. datetime/time：时间判断、过期清理、告警时间

import time

from datetime import datetime

# 时间戳转格式化时间

now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 7天前时间戳

seven_day = time.time() - 7 * 86400

14. configparser：读取配置文件（主机清单、阈值配置）

# ops.ini

[monitor]

disk_warn=90

cpu_warn=85

import configparser

cfg = configparser.ConfigParser()

cfg.read("ops.ini")

disk_threshold = int(cfg["monitor"]["disk_warn"])

五、Python 运维四大经典实战场景（生产脚本模板）

15. 场景1：服务器资源监控告警脚本（psutil+logging）

功能：CPU/内存/磁盘超过阈值写入告警日志，可对接钉钉/短信。

import psutil

import logging

# 阈值

CPU_THRESH = 85

MEM_THRESH = 90

DISK_THRESH = 90

def init_log():

    logger = logging.getLogger("monitor")

    logger.setLevel(logging.WARNING)

    handler = logging.FileHandler("/var/log/monitor.log", encoding="utf-8")

    fmt = logging.Formatter("%(asctime)s %(message)s")

    handler.setFormatter(fmt)

    logger.addHandler(handler)

    return logger

logger = init_log()

# CPU检测

cpu = psutil.cpu_percent(1)

if cpu > CPU_THRESH:

    logger.warning(f"CPU使用率过高：{cpu}%")

# 内存检测

mem = psutil.virtual_memory()

if mem.percent > MEM_THRESH:

    logger.warning(f"内存使用率过高：{mem.percent}%")

# 磁盘检测

for part in psutil.disk_partitions():

    if "/dev" not in part.device and part.mountpoint != "/tmp":

        usage = psutil.disk_usage(part.mountpoint)

        if usage.percent > DISK_THRESH:

            logger.warning(f"磁盘{part.mountpoint}使用率{usage.percent}%")

16. 场景2：日志自动清理脚本（定时crontab执行）

删除目录7天前日志文件，保留目录结构。

from pathlib import Path

import time

def clean_logs(log_dir, keep_day=7):

    expire = time.time() - keep_day * 86400

    p = Path(log_dir)

    if not p.exists():

        return

    for file in p.rglob("*.log"):

        if file.is_file():

            mtime = file.stat().st_mtime

            if mtime < expire:

                file.unlink()

clean_logs("/data/app/logs", 7)

配合 Linux crontab 每日凌晨执行：

0 0 * * * /usr/bin/python3 /opt/ops/clean_log.py

17. 场景3：批量远程服务器磁盘巡检（paramiko批量）

def batch_check_disk(host_list):

    result = []

    for info in host_list:

        ok, out, err = ssh_exec(info["host"], info["user"], info["pwd"], "df -h")

        result.append({

            "host": info["host"],

            "success": ok,

            "output": out,

            "error": err

        })

    return result

18. 场景4：进程巡检，杀死占用内存过高异常进程

import psutil

def kill_high_mem_proc(mem_limit=80):

    for proc in psutil.process_iter(["pid", "name", "memory_percent"]):

        try:

            info = proc.info

            pid = info["pid"]

            name = info["name"]

            mem_pct = info["memory_percent"]

            if mem_pct >= mem_limit:

                p = psutil.Process(pid)

                p.terminate()

                print(f"杀死高内存进程 pid:{pid} {name} 内存{mem_pct}%")

        except (psutil.NoSuchProcess, psutil.AccessDenied):

            continue

六、Python 运维对比 Shell 的优缺点

19. Python 优势

1. 原生数据结构：列表、字典、json，结构化处理输出，shell 文本切割极其繁琐；
2. 完善异常捕获 try-except，脚本崩溃可记录日志，Shell 错误极易静默失败；
3. 超大文件迭代读取不占内存，Shell 管道处理大日志容易卡顿；
4. 逻辑复杂：循环嵌套、多分支判断、阈值计算、批量远程操作可读性远高于Shell；
5. 标准日志库，自动切割、分级日志，Shell 只能简单重定向；
6. 跨主机SSH批量操作封装简单，Shell 批量需要expect、sshpass易出安全漏洞；
7. 内置正则、时间处理，不用依赖awk/sed/grep拼凑命令。

20. Shell 优势

1. 单行快速执行简单命令，无需写完整脚本；
2. 系统自带，无需安装 Python3；
3. 管道、文本过滤一行搞定简单查询。

21. 适用区分

• 简单临时查询、单行操作：Shell
• 自动化巡检、定时清理、批量运维、监控采集、日志分析、告警平台对接：Python

七、线上运维脚本标准规范

1. 头部写清功能、入参、执行周期；
2. 所有文件操作、命令执行加 try-except 捕获异常；
3. 全部使用 logging 落地日志，禁止大量print；
4. 超大文件统一迭代器逐行读取，禁止 read() 全量加载；
5. 远程SSH操作加超时，防止脚本卡死；
6. 资源操作（杀进程、删日志）增加判断，避免误删；
7. 脚本交给 crontab 定时执行，输出重定向日志；
8. 配置与代码分离，阈值、主机列表放入ini/json配置文件；
9. 敏感密码禁止硬编码，使用环境变量、密钥登录。

八、面试核心速记总结

1. Python运维四大基础库：subprocess执行shell、psutil采集系统指标、os/shutil文件目录操作、logging日志；
2. subprocess优先列表传参防注入，capture_output捕获输出，timeout防卡死；
3. psutil无需调用命令，直接获取CPU/内存/磁盘/网络/进程，写监控脚本核心；
4. 文件处理超大日志用 for line in f 迭代器逐行读取，内存恒定；
5. paramiko 实现SSH远程批量执行、文件上传下载，批量运维必备；
6. 定时任务结合Linux crontab 执行Python清理/巡检脚本；
7. Python相比Shell优势：异常处理、结构化数据、大文件低内存、批量远程、规范日志；
8. 运维高频场景：资源监控告警、日志过期清理、批量服务器巡检、异常进程查杀、日志错误过滤。

关联文档

• Python 文件操作完整详解
• Python 基础
• Shell 基础