运维脚本实战项目

Python 的语法、标准库和系统交互分散在不同篇章里，缺一个把它们串起来的完整例子。ops-http-checker 就是这样的小项目——一个 HTTP 健康检查工具，读取目标配置、并发检查一批 URL、生成 JSON 报告、失败时返回非零退出码，能直接挂到 cron 或监控系统上跑。

项目拆成几个文件，每个文件管一摊：配置读取、检查逻辑、报告生成、入口。每读一个文件，先看它解决什么问题，再看代码——不要一上来就扎进代码细节。

一、项目结构

text

ops-http-checker/
├── configs/
│   └── targets.json       # 检查目标配置
├── reports/               # 报告输出目录
├── src/
│   └── ops_http_checker/
│       ├── __init__.py
│       ├── checker.py     # 检查逻辑(类)
│       ├── config.py      # 配置读取
│       ├── report.py      # 报告生成
│       └── cli.py         # 入口
└── pyproject.toml

初始化:

bash

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

src/ops_http_checker/ 是一个包,__init__.py 标记它能被 import。每个模块按职责拆分——config.py 读配置、checker.py 做检查、report.py 生成报告、cli.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_http_checker/config.py

python

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

import json
from pathlib import Path


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


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),  # 没配就用默认 5 秒
        })

    return targets

配置读取用到了自定义异常(ConfigError)、pathlib、JSON 解析、字段校验。校验在启动时就做完,别等跑到一半才发现某个目标缺 URL——那时已经检查了一半,结果难看。

四、检查模块

新增文件:src/ops_http_checker/checker.py

python

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

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 是个类，把检查结果的数据（名字、URL、是否成功、消息、耗时）打包在一起。为什么用类不用字典？ 因为类能带方法——to_dict() 把对象转成字典写报告，后面如果想加 is_failed() 这种判断方法也方便。字典做不到附带方法。如果只是存数据、不需要方法，用字典也行，这里用类是为了让结果对象能自带行为。

batch_check 用线程池并发,单个目标失败不中断整批——错误变成结果的一部分。

五、报告模块

新增文件:src/ops_http_checker/report.py

python

"""生成检查报告。"""

import json
from datetime import datetime
from pathlib import Path


def generate_report(results, output_dir):
    """生成 JSON 报告,返回报告文件路径。"""
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    report_path = output_dir / f"report-{timestamp}.json"

    summary = {
        "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "total": len(results),
        "passed": sum(1 for r in results if r.ok),
        "failed": sum(1 for r in results if not r.ok),
        "results": [r.to_dict() for r in results],
    }

    report_path.write_text(
        json.dumps(summary, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    return report_path


def print_summary(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 = "✓" if r.ok else "✗"
        print(f"  {status} {r.name} ({r.cost_ms}ms) {r.message}")

报告用 JSON 格式,带时间戳文件名——每次运行生成一份,不覆盖历史报告。datetime.now().strftime 给文件名加日期,跟文件与文本处理里讲的一样。

六、入口

新增文件:src/ops_http_checker/cli.py

python

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

import argparse
import logging
import sys

from ops_http_checker.config import load_targets, ConfigError
from ops_http_checker.checker import batch_check
from ops_http_checker.report import generate_report, print_summary


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("--report-dir", default="reports", help="报告输出目录")
    return parser.parse_args()


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

    # 1. 读配置
    try:
        targets = load_targets(args.config)
    except ConfigError as exc:
        logging.error("配置错误: %s", exc)
        return 2  # 配置错误用 2,跟检查失败(1)区分

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

    # 2. 并发检查
    results = batch_check(targets, workers=args.workers)

    # 3. 打印摘要 + 生成报告
    print_summary(results)
    report_path = generate_report(results, args.report_dir)
    logging.info("报告已生成: %s", report_path)

    # 4. 有失败就返回非零,cron/监控能识别
    return 1 if any(not r.ok for r in results) else 0


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

入口的流程就四步:读配置 → 并发检查 → 输出报告 → 返回退出码。每一步都用前面讲过的东西——配置用 config 模块,检查用 checker 模块,报告用 report 模块,argparse 接参数,logging 记日志。退出码区分了三种情况:0 全部成功、1 有检查失败、2 配置错误——监控系统能根据退出码判断是"服务挂了"还是"脚本本身配错了"。

七、运行

bash

# 安装项目依赖
uv sync

# 运行检查
# -m 表示把后面的名字当模块跑，python 会在 sys.path 里找 ops_http_checker 包，
# 再执行里面的 cli.py
uv run python -m ops_http_checker.cli --config configs/targets.json --workers 5

# 看退出码
echo $?

输出大概长这样:

text

2024-06-01 10:30:00 INFO 加载了 2 个检查目标
总计 2 个目标,成功 2,失败 0
  ✓ api (152ms) status_code=200
  ✓ grafana (89ms) status_code=200
2024-06-01 10:30:00 INFO 报告已生成: reports/report-20240601-103000.json

报告文件 reports/report-20240601-103000.json:

json

{
  "generated_at": "2024-06-01 10:30:00",
  "total": 2,
  "passed": 2,
  "failed": 0,
  "results": [
    {"name": "api", "url": "https://example.com/health", "ok": true, "message": "status_code=200", "cost_ms": 152},
    {"name": "grafana", "url": "http://127.0.0.1:3000/api/health", "ok": true, "message": "status_code=200", "cost_ms": 89}
  ]
}

八、挂到 cron 定时跑

cron

# 每 5 分钟跑一次健康检查
*/5 * * * * cd /path/to/ops-http-checker && uv run python -m ops_http_checker.cli >> /var/log/ops-http-checker.log 2>&1

cron 里要注意几点:cd 到项目目录(不然相对路径找不到配置),用 uv run 而不是 python(确保用项目虚拟环境),输出重定向到日志文件(cron 默认不输出到终端)。这些跟系统交互里讲 cron 的坑是同一套。

九、之后能往哪扩展

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

重试:检查失败时重试几次,在 check_one 里加循环(参考进阶语法里的装饰器重试)
告警通知:失败时发企业微信/钉钉,在 main 里检查 failed > 0 后调 webhook
历史趋势:把每次报告存进数据库,画一段时间内的可用率曲线
配置热更新:改 targets.json 后不用重启,下次检查自动读到新配置(每次 load_targets 都读文件就行)

每加一个功能,骨架不变,只改对应的模块。这就是前面花力气学函数、类、模块、异常、文件处理的意义——它们组合起来,能搭出真正能用的工具。

运维脚本实战项目 ​

一、项目结构 ​

二、配置文件 ​

三、配置读取模块 ​

四、检查模块 ​

五、报告模块 ​

六、入口 ​

七、运行 ​

八、挂到 cron 定时跑 ​

九、之后能往哪扩展 ​

运维脚本实战项目

一、项目结构

二、配置文件

三、配置读取模块

四、检查模块

五、报告模块

六、入口

七、运行

八、挂到 cron 定时跑

九、之后能往哪扩展