系统交互

前面几篇都在 Python 语言内部打转。从这篇开始正式接触运维场景——执行系统命令、读环境变量、记日志、解析命令行参数。Shell 调命令很直接,但 Python 的优势在于能接住返回码、标准输出、错误输出,再做结构化处理;涉及多步骤、异常分支和结果汇总时,Python 比 Shell 好维护得多。

一、环境变量

环境变量常用来传账号、URL、环境名、开关这些不写死在代码里的值:

import os

env = os.environ.get("APP_ENV", "dev")  # 没设过就用默认值 dev
api_url = os.environ.get("API_URL")      # 没设过返回 None

print(f"env={env} api_url={api_url}")

必填的环境变量集中校验,启动时缺了直接报错,别等跑到一半才发现:

import os


def require_env(name):
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"missing environment variable: {name}")
    return value


token = require_env("API_TOKEN")

密码、Token 这类敏感信息硬编码在脚本里是大忌——代码进版本库、发到聊天群,泄露面一下子就大了。环境变量、配置文件(权限收紧)、密钥管理系统都比硬编码可控。

二、执行系统命令

subprocess.run() 是执行外部命令的标准入口:

import subprocess

result = subprocess.run(
    ["systemctl", "is-active", "nginx"],
    text=True,          # 输出按字符串处理,不返回 bytes
    capture_output=True, # 捕获 stdout 和 stderr
    check=False,         # 非零退出码不自动抛异常,脚本自己判断
)

print(result.returncode)     # 0 成功,非 0 失败
print(result.stdout.strip()) # 正常输出
print(result.stderr.strip()) # 错误输出

命令和参数用列表传——["systemctl", "is-active", "nginx"],不要拼成字符串。这是为了避开 shell=True 的安全坑:

# 危险写法:shell=True + 字符串拼接
host = "; rm -rf /tmp"  # 假设 host 来自外部输入
subprocess.run(f"ping -c 1 {host}", shell=True)
# Shell 会把这一行解析成:先 ping,再执行 rm -rf /tmp

# 安全写法:列表传参
subprocess.run(["ping", "-c", "1", host])
# host 只是一个字符串参数,不会被当命令执行

shell=True 把字符串交给 Shell 解释。字符串里拼了外部输入,Shell 就可能把它当成命令执行——这就是命令注入。外部输入永远用列表传参,别用 shell=True 拼字符串。

封装一个通用的执行函数,带上超时:

import subprocess


def run_cmd(args, timeout=30):
    """执行命令,返回 CompletedProcess。"""
    return subprocess.run(
        args,
        text=True,
        capture_output=True,
        timeout=timeout,  # 超时杀掉,避免卡死
        check=False,
    )


result = run_cmd(["df", "-h"])
if result.returncode != 0:
    print(f"failed: {result.stderr.strip()}")
else:
    print(result.stdout)

三、解析命令输出

命令输出是纯文本,解析时找稳定字段。能让命令输出机器可读格式时,优先选机器可读的参数:

import subprocess


def get_disk_usage():
    result = subprocess.run(
        ["df", "-P"],   # -P 用 POSIX 格式,输出稳定,适合脚本解析
        text=True,
        capture_output=True,
        check=True,
    )

    rows = []
    for line in result.stdout.splitlines()[1:]:  # 跳过表头
        parts = line.split(maxsplit=5)
        if len(parts) < 6:
            continue
        filesystem, blocks, used, available, use_percent, mountpoint = parts
        rows.append({
            "mountpoint": mountpoint,
            "used_percent": int(use_percent.rstrip("%")),
        })
    return rows


for item in get_disk_usage():
    if item["used_percent"] >= 80:
        print(f"{item['mountpoint']} used={item['used_percent']}%")

df -P 比 df -h 更适合脚本——-h 输出带 G、M 单位,解析要处理单位;-P 输出纯数字,字段位置固定。能用稳定参数就别用"给人看"的参数。

四、日志

print() 适合一次性小脚本。稍微正式点的脚本用 logging,能区分级别、带时间、还能写文件:

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

logging.info("script started")
logging.warning("disk usage high")
logging.error("service check failed")

写文件:

import logging
from pathlib import Path

logging.basicConfig(
    filename=Path("/var/log/ops-script.log"),
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

logging.info("write log to file")

日志要写清楚对象和结果,别只写"出错了":

# 含糊
logging.error("check failed")

# 清楚:谁、做了什么、结果是什么
logging.info("check_service host=%s service=%s status=%s", "web01", "nginx", "active")

logging 用 %s 占位符而不是 f-string——日志级别被过滤时(比如设成 WARNING,INFO 不输出),占位符不会被格式化,省开销。

五、配置文件

1 ini 配置

简单的键值配置用 configparser:

# /etc/ops/check.ini
[api]
url = https://example.com/health
timeout = 5

[check]
threshold = 80

import configparser
from pathlib import Path

config = configparser.ConfigParser()
config.read(Path("/etc/ops/check.ini"), encoding="utf-8")

api_url = config["api"]["url"]
timeout = config.getint("api", "timeout")   # getint 自动转整数
threshold = config.getint("check", "threshold")

2 YAML 配置

YAML 支持嵌套和列表,比 ini 灵活,但需要第三方包 pyyaml:

uv add pyyaml

# targets.yaml
targets:
  - name: nginx
    host: 127.0.0.1
    port: 80
  - name: mysql
    host: 127.0.0.1
    port: 3306

from pathlib import Path
import yaml

config = yaml.safe_load(Path("targets.yaml").read_text(encoding="utf-8"))
for target in config["targets"]:
    print(target["name"], target["host"], target["port"])

safe_load 比 load 安全——load 能反序列化任意 Python 对象,读到恶意的 YAML 会执行代码,safe_load 只解析基础数据类型。

六、临时文件和锁

临时文件用 tempfile,系统会分配唯一路径,不用自己造名字:

import tempfile

with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False) as f:
    f.write("temporary data\n")
    print(f.name)  # 临时文件路径

delete=False 表示退出 with 块后不自动删,方便排查;默认 delete=True 会自动清理。

防止同一个定时脚本重复运行,用锁文件:

from pathlib import Path
import os

lock_path = Path("/tmp/check-service.lock")

if lock_path.exists():
    raise SystemExit(f"lock exists: {lock_path}, 上一次可能还在跑")

try:
    lock_path.write_text(str(os.getpid()), encoding="utf-8")  # 写 PID 方便排查
    print("running job")
finally:
    lock_path.unlink(missing_ok=True)  # 不管成功失败都删锁

这种锁文件适合单机简单场景。跨机器、长任务、异常退出后自动清理这些需求,要换 Redis 锁或调度系统。

七、命令行参数

argparse 解析命令行参数,自动生成帮助信息:

import argparse


def parse_args():
    parser = argparse.ArgumentParser(description="check service port")
    parser.add_argument("--host", required=True, help="target host")
    parser.add_argument("--port", required=True, type=int, help="target port")
    parser.add_argument("--timeout", type=int, default=3, help="connect timeout seconds")
    return parser.parse_args()


args = parse_args()
print(args.host, args.port, args.timeout)

运行:

uv run python check_port.py --host 127.0.0.1 --port 22 --timeout 2

# 自动生成帮助
uv run python check_port.py --help

主机、端口、阈值、配置路径这些会变的值,写成参数或配置后,脚本复用时不用改源码。

八、实战:systemd 状态检查

把参数、命令执行、日志、退出码合到一个完整脚本:

新增文件:scripts/check_systemd.py

#!/usr/bin/env python3
"""检查 systemd 服务状态。"""

import argparse
import logging
import subprocess
import sys


def parse_args():
    parser = argparse.ArgumentParser(description="check systemd service state")
    parser.add_argument("service", help="systemd service name, e.g. nginx")
    return parser.parse_args()


def get_service_state(service):
    result = subprocess.run(
        ["systemctl", "is-active", service],
        text=True,
        capture_output=True,
        check=False,
    )
    return result.returncode, result.stdout.strip(), result.stderr.strip()


def main():
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    args = parse_args()

    returncode, stdout, stderr = get_service_state(args.service)
    if returncode == 0 and stdout == "active":
        logging.info("service=%s state=active", args.service)
        return 0

    logging.error("service=%s state=%s error=%s", args.service, stdout, stderr)
    return 1


if __name__ == "__main__":
    sys.exit(main())

运行:

uv run python scripts/check_systemd.py nginx
echo $?  # 0 正常,1 异常

这个脚本结构是运维脚本的通用骨架——参数解析、执行检查、记日志、返回退出码。后面加配置文件、批量目标、生成报告,都是在这个骨架上扩展,入口不用推倒重写。