Python Linux 系统运维详解
核心概念 Python 适合复杂、可维护、批量化的自动化运维脚本，覆盖服务器信息采集、进程管理、网络监控、日志分析等。
1. Python 运维核心定位
Shell 适合单行简单命令；Python 适合复杂、可维护、批量、有逻辑、带日志、异常处理、结构化输出的自动化运维脚本。
核心覆盖：
服务器信息采集、文件/目录管理、进程管理、磁盘/内存/网络监控、定时任务、批量远程主机操作、日志分析、服务启停、系统参数调优、容器基础操作。
运维必备四大标准库：
os：底层系统、目录、文件、进程基础操作shutil：高级文件复制、移动、递归删除（os 增强）subprocess：执行 Linux shell 命令，采集命令输出（运维最常用）psutil：跨平台系统资源采集（CPU/内存/磁盘/网络/进程，运维神器）
扩展高频库：
paramiko：SSH 远程批量操作多台 Linuxconfigparser/tomli：读取配置文件logging：脚本日志落地，线上必备datetime/time：时间戳、定时判断、告警时间re：日志正则匹配、过滤报错、提取指标json：结构化指标输出，对接监控平台
二、四大核心运维标准库深度讲解
2. subprocess：执行 Linux 命令（运维第一核心）
所有查看系统状态、启停服务、df/top/netstat/curl 全部依靠它。
2.1. 核心规则
shell=False：列表传参，无注入风险，推荐；shell=True：支持管道|、通配符*，不可接收外部用户输入；capture_output=True捕获 stdout/stderr；text=True输出字符串，避免 bytes 解码；timeout防止命令卡死阻塞脚本；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 psutil3.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_bytes3.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_lines5.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 paramiko8. 基础单台主机执行命令
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 * 8640014. configparser：读取配置文件（主机清单、阈值配置）
# ops.ini
[monitor]
disk_warn=90
cpu_warn=85import 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.py17. 场景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 result18. 场景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 优势
- 原生数据结构：列表、字典、json，结构化处理输出，shell 文本切割极其繁琐；
- 完善异常捕获
try-except，脚本崩溃可记录日志，Shell 错误极易静默失败； - 超大文件迭代读取不占内存，Shell 管道处理大日志容易卡顿；
- 逻辑复杂：循环嵌套、多分支判断、阈值计算、批量远程操作可读性远高于Shell；
- 标准日志库，自动切割、分级日志，Shell 只能简单重定向；
- 跨主机SSH批量操作封装简单，Shell 批量需要expect、sshpass易出安全漏洞；
- 内置正则、时间处理，不用依赖awk/sed/grep拼凑命令。
20. Shell 优势
- 单行快速执行简单命令，无需写完整脚本；
- 系统自带，无需安装 Python3；
- 管道、文本过滤一行搞定简单查询。
21. 适用区分
- 简单临时查询、单行操作：Shell
- 自动化巡检、定时清理、批量运维、监控采集、日志分析、告警平台对接：Python
七、线上运维脚本标准规范
- 头部写清功能、入参、执行周期；
- 所有文件操作、命令执行加
try-except捕获异常； - 全部使用
logging落地日志，禁止大量print； - 超大文件统一迭代器逐行读取，禁止
read()全量加载； - 远程SSH操作加超时，防止脚本卡死；
- 资源操作（杀进程、删日志）增加判断，避免误删；
- 脚本交给 crontab 定时执行，输出重定向日志；
- 配置与代码分离，阈值、主机列表放入ini/json配置文件；
- 敏感密码禁止硬编码，使用环境变量、密钥登录。
八、面试核心速记总结
- Python运维四大基础库：subprocess执行shell、psutil采集系统指标、os/shutil文件目录操作、logging日志；
- subprocess优先列表传参防注入，capture_output捕获输出，timeout防卡死；
- psutil无需调用命令，直接获取CPU/内存/磁盘/网络/进程，写监控脚本核心；
- 文件处理超大日志用
for line in f迭代器逐行读取，内存恒定； - paramiko 实现SSH远程批量执行、文件上传下载，批量运维必备；
- 定时任务结合Linux crontab 执行Python清理/巡检脚本；
- Python相比Shell优势：异常处理、结构化数据、大文件低内存、批量远程、规范日志；
- 运维高频场景：资源监控告警、日志过期清理、批量服务器巡检、异常进程查杀、日志错误过滤。
关联文档