38｜运维脚本项目

前面 37 讲覆盖了 Python 的基础语法、数据结构、函数、类、文件处理、网络请求、并发和系统交互。现在把这些能力合成一个完整的运维脚本项目：批量 HTTP 健康检查工具 ops-checker。

它从配置文件读取目标列表，并发检查每个目标的 HTTP 状态，输出文本或 JSON 结果，失败时返回非零退出码。这个工具可以直接挂到 cron 或监控系统上跑。

一、项目结构

text

ops-checker/
├── pyproject.toml
├── configs/
│   └── targets.json
├── src/
│   └── ops_checker/
│       ├── __init__.py
│       ├── config.py      # 配置读取
│       ├── checker.py     # 检查逻辑
│       └── cli.py         # 入口
└── README.md

pyproject.toml 记录项目和依赖。src/ops_checker/ 是一个 Python 包，每个模块按职责拆分。

初始化项目：

bash

uv init ops-checker
cd ops-checker
uv add requests
mkdir -p configs src/ops_checker
touch src/ops_checker/__init__.py

二、配置文件

configs/targets.json：

json

[
  {
    "name": "api",
    "url": "https://example.com/health",
    "timeout": 3
  },
  {
    "name": "grafana",
    "url": "http://127.0.0.1:3000/api/health",
    "timeout": 2
  }
]

每个目标有名字、URL、超时。名字用在报告和日志里，超时单独配，因为不同服务的响应速度差别很大。

三、配置读取

src/ops_checker/config.py：

python

"""配置读取和校验。"""

import json
from pathlib import Path


class ConfigError(Exception):
    """配置文件格式不对。"""


def load_targets(path):
    """从 JSON 文件读取检查目标，校验必填字段。"""
    config_path = Path(path)

    if not config_path.exists():
        raise ConfigError(f"配置文件不存在: {config_path}")

    try:
        data = json.loads(config_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise ConfigError(f"配置文件 JSON 格式错误: {exc}") from exc

    if not isinstance(data, list):
        raise ConfigError("配置文件顶层应该是列表")

    targets = []
    for index, item in enumerate(data):
        for field in ("name", "url"):
            if field not in item:
                raise ConfigError(f"第 {index + 1} 个目标缺少字段: {field}")
        targets.append({
            "name": item["name"],
            "url": item["url"],
            "timeout": item.get("timeout", 5),
        })

    return targets

校验在启动时就做完，别等跑到一半才发现某个目标缺 URL。

四、检查逻辑

src/ops_checker/checker.py：

python

"""HTTP 健康检查。"""

import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests


class CheckResult:
    """单个目标的检查结果。"""

    def __init__(self, name, url, ok, message, cost_ms):
        self.name = name
        self.url = url
        self.ok = ok
        self.message = message
        self.cost_ms = cost_ms

    def to_dict(self):
        return {
            "name": self.name,
            "url": self.url,
            "ok": self.ok,
            "message": self.message,
            "cost_ms": self.cost_ms,
        }


def check_one(target):
    """检查单个目标，返回 CheckResult。"""
    name = target["name"]
    url = target["url"]
    timeout = target["timeout"]

    started = time.time()
    try:
        response = requests.get(url, timeout=timeout)
        ok = response.status_code == 200
        message = f"status_code={response.status_code}"
    except requests.RequestException as exc:
        ok = False
        message = str(exc)

    return CheckResult(
        name=name,
        url=url,
        ok=ok,
        message=message,
        cost_ms=int((time.time() - started) * 1000),
    )


def batch_check(targets, workers=5):
    """并发检查一批目标，返回 CheckResult 列表。"""
    results = []

    with ThreadPoolExecutor(max_workers=workers) as executor:
        future_map = {executor.submit(check_one, t): t for t in targets}
        for future in as_completed(future_map):
            try:
                results.append(future.result())
            except Exception as exc:
                target = future_map[future]
                results.append(CheckResult(
                    name=target["name"],
                    url=target["url"],
                    ok=False,
                    message=f"unexpected error: {exc}",
                    cost_ms=0,
                ))

    return results

CheckResult 用类封装检查结果，比字典更适合附带方法（如 to_dict()）。batch_check 用线程池并发，单个目标失败不中断整批——错误变成结果的一部分。

五、入口

src/ops_checker/cli.py：

python

"""命令行入口。"""

import argparse
import json
import logging
import sys

from ops_checker.config import ConfigError, load_targets
from ops_checker.checker import batch_check


def parse_args():
    parser = argparse.ArgumentParser(description="HTTP 健康检查工具")
    parser.add_argument("--config", default="configs/targets.json", help="目标配置文件")
    parser.add_argument("--workers", type=int, default=5, help="并发数")
    parser.add_argument("--output", choices=["text", "json"], default="text", help="输出格式")
    return parser.parse_args()


def print_text(results):
    total = len(results)
    passed = sum(1 for r in results if r.ok)
    failed = total - passed

    print(f"总计 {total} 个目标，成功 {passed}，失败 {failed}")
    for r in results:
        status = "OK" if r.ok else "FAIL"
        print(f"  [{status}] {r.name} ({r.cost_ms}ms) {r.message}")


def print_json(results):
    data = [r.to_dict() for r in results]
    print(json.dumps(data, ensure_ascii=False, indent=2))


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

    try:
        targets = load_targets(args.config)
    except ConfigError as exc:
        logging.error("配置错误: %s", exc)
        return 2

    logging.info("加载了 %d 个检查目标", len(targets))

    results = batch_check(targets, workers=args.workers)

    if args.output == "json":
        print_json(results)
    else:
        print_text(results)

    return 1 if any(not r.ok for r in results) else 0


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

入口流程：解析参数 → 读取配置 → 并发检查 → 输出结果 → 返回退出码。

退出码含义：

退出码	含义
0	所有目标检查成功
1	有目标检查失败
2	配置错误

六、运行

bash

# 安装依赖
uv sync

# 运行检查
uv run python -m ops_checker.cli --config configs/targets.json

# JSON 输出
uv run python -m ops_checker.cli --output json

# 看退出码
echo $?

七、挂到 cron

cron

*/5 * * * * cd /path/to/ops-checker && uv run python -m ops_checker.cli >> /var/log/ops-checker.log 2>&1

cron 里要注意：cd 到项目目录（不然相对路径找不到配置），用 uv run 而不是 python（确保用项目虚拟环境），输出重定向到日志文件。

八、扩展方向

这个项目是个骨架，后面加功能都是在它上面叠：

重试：检查失败时重试几次，在 check_one 里加循环
告警通知：失败时发企业微信/钉钉，在 main 里检查 failed > 0 后调 webhook
历史趋势：把每次报告存进数据库，画一段时间内的可用率曲线
更多检查类型：TCP 端口检查、证书过期检查，新增检查函数并扩展配置格式

每加一个功能，骨架不变，只改对应的模块。

记忆锚点：ops-checker 层次：配置 → 检查 → 结果 → 输出 → 退出码；配置校验在启动时做完；线程池并发，单目标失败不中断整批；退出码 0/1/2 分别表示成功/目标失败/配置错误；cron 里 cd 到项目目录，用 uv run；项目是骨架，新增功能只改对应模块。

38｜运维脚本项目 ​

一、项目结构 ​

二、配置文件 ​

三、配置读取 ​

四、检查逻辑 ​

五、入口 ​

六、运行 ​

七、挂到 cron ​

八、扩展方向 ​

38｜运维脚本项目

一、项目结构

二、配置文件

三、配置读取

四、检查逻辑

五、入口

六、运行

七、挂到 cron

八、扩展方向