运维脚本模板库

本篇收集写运维 Python 脚本时最常复用的骨架片段，每段都能独立运行。复制后改掉路径、阈值等参数即可使用。片段均按 Python 3.12+ 编写，第三方库在片段处注明安装命令。

脚本标准骨架

新脚本的标准起手式：shebang + argparse 参数解析 + main 入口，自带 --dry-run 是好习惯。

#!/usr/bin/env python3
"""一句话说明脚本做什么"""

import argparse
import sys


def main():
    parser = argparse.ArgumentParser(description="一句话说明脚本做什么")
    parser.add_argument("target", help="目标主机或文件")
    parser.add_argument("-n", "--dry-run", action="store_true", help="只打印不执行")
    parser.add_argument("-v", "--verbose", action="store_true", help="输出调试信息")
    args = parser.parse_args()

    if args.dry_run:
        print(f"[dry-run] 将对 {args.target} 执行操作")
        return 0
    # 真正的逻辑写在这里
    return 0


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

logging 控制台与文件双输出

控制台看简洁版、文件里留带时间戳的完整版，进 cron 之后全靠日志文件排查问题。

import logging


def setup_logging(log_file="app.log", level=logging.INFO):
    """控制台 + 文件双输出：控制台简洁，文件带时间戳"""
    logger = logging.getLogger()
    logger.setLevel(level)

    console = logging.StreamHandler()
    console.setFormatter(logging.Formatter("%(levelname)s %(message)s"))

    file_handler = logging.FileHandler(log_file, encoding="utf-8")
    file_handler.setFormatter(
        logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")
    )

    logger.addHandler(console)
    logger.addHandler(file_handler)
    return logger


if __name__ == "__main__":
    log = setup_logging("demo.log")
    log.info("脚本启动")
    log.warning("这是一条警告")

重复调用会叠加 handler

setup_logging 在一个进程里只调一次。重复调用会重复 addHandler，一条日志打好几遍——模块里建议只做 logging.getLogger(__name__)，配置入口集中在一处。

subprocess 执行命令带超时和重试

调外部命令的封装：超时强杀、失败自动重试、返回码和输出一起拿回来。

import subprocess
import time


def run_cmd(cmd, timeout=30, retries=3, interval=2):
    """执行命令，超时或失败自动重试，返回 (成功与否, 输出或错误)"""
    err = ""
    for attempt in range(1, retries + 1):
        try:
            result = subprocess.run(
                cmd, shell=True, capture_output=True, text=True, timeout=timeout
            )
            if result.returncode == 0:
                return True, result.stdout.strip()
            err = result.stderr.strip()
        except subprocess.TimeoutExpired:
            err = f"超过 {timeout}s 未返回"
        if attempt < retries:
            time.sleep(interval)
    return False, f"重试 {retries} 次仍失败，最后错误: {err}"


if __name__ == "__main__":
    ok, output = run_cmd("systemctl is-active nginx")
    print(ok, output)

读取 yaml 配置

配置外置后的标准读法，yaml.safe_load 比 load 安全，文件不存在时给出能看懂的报错。依赖：python3 -m pip install PyYAML。

from pathlib import Path

import yaml


def load_config(path="config.yaml"):
    """读取 yaml 配置，文件不存在时给出清晰报错"""
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"配置文件不存在: {p.resolve()}")
    with p.open(encoding="utf-8") as f:
        return yaml.safe_load(f)


if __name__ == "__main__":
    conf = load_config()
    print(conf["threshold"])

配套的配置文件长这样：

# config.yaml
threshold: 85
hosts:
  - 192.168.1.11
  - 192.168.1.12

retry 重试装饰器

网络请求、SSH 连接这类「抖一下就好」的操作，套个装饰器就有了重试能力，不用改函数本身。

import functools
import time


def retry(times=3, interval=2, exceptions=(Exception,)):
    """重试装饰器：捕获指定异常后间隔重试，耗尽后抛出最后一次异常"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions:
                    if attempt == times:
                        raise
                    time.sleep(interval)
        return wrapper
    return decorator


if __name__ == "__main__":
    import requests  # python3 -m pip install requests

    @retry(times=3, interval=5, exceptions=(requests.RequestException,))
    def fetch(url):
        return requests.get(url, timeout=10).json()

    print(fetch("https://api.github.com"))

ThreadPoolExecutor 并发池模板

批量操作多台机器/多个文件的通用并发骨架：控制并发上限，单个失败不影响整体，结果分成成功和失败两堆。

from concurrent.futures import ThreadPoolExecutor, as_completed


def handle(item):
    """单个对象的操作，异常向上抛，由 run_pool 收集"""
    return f"{item} done"


def run_pool(items, workers=10):
    """并发执行并汇总结果，单个失败不拖垮整体"""
    results = {"ok": [], "fail": []}
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(handle, item): item for item in items}
        for future in as_completed(futures):
            item = futures[future]
            try:
                results["ok"].append((item, future.result()))
            except Exception as exc:
                results["fail"].append((item, str(exc)))
    return results


if __name__ == "__main__":
    hosts = [f"192.168.1.{i}" for i in range(11, 21)]
    r = run_pool(hosts)
    print(f"成功 {len(r['ok'])} 个，失败 {len(r['fail'])} 个")

读取环境变量带默认值

密钥、开关这类运行期才确定的值走环境变量，代码里留默认值兜底；必需但没给的在启动时就报错退出，别跑到一半才炸。

import os

WEBHOOK = os.environ.get("ROBOT_WEBHOOK", "")              # 字符串，默认空
THRESHOLD = int(os.environ.get("DISK_THRESHOLD", "85"))    # 数值，记得转类型
DEBUG = os.environ.get("DEBUG", "").lower() in ("1", "true", "yes")  # 布尔开关

if not WEBHOOK:
    raise SystemExit("缺少环境变量 ROBOT_WEBHOOK，先 export 再运行")

布尔环境变量的坑

os.environ.get("DEBUG", "False") 拿到的是字符串，非空字符串在 if 里都是 True——设成 DEBUG=False 反而开启了调试。所以布尔值要像上面那样显式比对。

原子写文件

脚本写配置或状态文件时，先写临时文件再 os.replace 替换，中途断电、Ctrl+C 都不会留下写了一半的坏文件。

import os
import tempfile
from pathlib import Path


def atomic_write(path, content):
    """先写同目录临时文件再 rename 覆盖，避免半截文件"""
    path = Path(path)
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".tmp-")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
        os.replace(tmp, path)
    except BaseException:
        os.unlink(tmp)
        raise


if __name__ == "__main__":
    atomic_write("/tmp/state.json", '{"last_alert": 0}')

临时文件要和目标同目录

os.replace 依赖 rename 系统调用，跨文件系统会失败。临时文件必须建在目标文件的同目录（上面的 dir=path.parent），不能图省事放 /tmp。