用 Python 检查定时任务产物是否过期：修改时间、空文件与退出码

python3 --version
mkdir freshness-demo
cd freshness-demo
freshness_root="$PWD"
python3 - <<'PY_DATA'
from pathlib import Path
import os
import time

now = time.time()
for name, content, modified in [
    ("fresh.txt", "ready\n", now),
    ("stale.txt", "ready\n", now - 7200),
    ("empty.txt", "", now),
    ("future.txt", "ready\n", now + 3600),
]:
    Path(name).write_text(content, encoding="utf-8")
    os.utime(name, (modified, modified))
PY_DATA

"""Check a completed output file's age and size. Python 3.9+, standard library."""
import argparse
import json
from pathlib import Path
import stat
import sys
import time


def nonnegative_int(value):
    number = int(value)
    if number < 0:
        raise argparse.ArgumentTypeError("must be zero or greater")
    return number


def inspect_file(path, max_age, min_bytes):
    info = path.stat()
    if not stat.S_ISREG(info.st_mode):
        raise ValueError("not a regular file")
    age = time.time() - info.st_mtime
    problems = []
    if info.st_size < min_bytes:
        problems.append("too_small")
    if age < -5:
        problems.append("future_mtime")
    elif age > max_age:
        problems.append("too_old")
    return {
        "file": str(path),
        "status": "alert" if problems else "ok",
        "age_seconds": round(age, 3),
        "size_bytes": info.st_size,
        "problems": problems,
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("file", type=Path, help="completed output to inspect")
    parser.add_argument("--max-age", type=nonnegative_int, required=True,
                        help="maximum age in seconds; must be greater than zero")
    parser.add_argument("--min-bytes", type=nonnegative_int, default=1,
                        help="minimum size in bytes (default: 1)")
    args = parser.parse_args()
    if args.max_age == 0:
        parser.error("--max-age must be greater than zero")
    try:
        result = inspect_file(args.file, args.max_age, args.min_bytes)
    except (OSError, ValueError) as exc:
        print(f"ERROR: {args.file}: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, ensure_ascii=False))
    return 1 if result["problems"] else 0


if __name__ == "__main__":
    raise SystemExit(main())

python3 check_freshness.py --help
python3 check_freshness.py fresh.txt --max-age 3600
python3 check_freshness.py stale.txt --max-age 3600

{"file": "fresh.txt", "status": "ok", "age_seconds": 2.314, "size_bytes": 6, "problems": []}

python3 - <<'PY_CHECK'
import json
import subprocess
import sys

cases = [
    (["fresh.txt", "--max-age", "3600"], 0, []),
    (["stale.txt", "--max-age", "3600"], 1, ["too_old"]),
    (["empty.txt", "--max-age", "3600"], 1, ["too_small"]),
    (["future.txt", "--max-age", "3600"], 1, ["future_mtime"]),
    (["missing.txt", "--max-age", "3600"], 1, None),
    ([".", "--max-age", "3600"], 1, None),
    (["fresh.txt", "--max-age", "0"], 2, None),
]
for arguments, expected_code, expected_problems in cases:
    result = subprocess.run(
        [sys.executable, "check_freshness.py", *arguments],
        capture_output=True, text=True, check=False,
    )
    assert result.returncode == expected_code, (arguments, result)
    if expected_problems is None:
        assert result.stdout == "" and result.stderr, (arguments, result)
    else:
        report = json.loads(result.stdout)
        assert report["problems"] == expected_problems, report
        assert report["status"] == ("alert" if expected_problems else "ok"), report
print("OK: fresh, stale, empty, future, missing, directory, bad arguments")
PY_CHECK

OK: fresh, stale, empty, future, missing, directory, bad arguments

python3 "$freshness_root/check_freshness.py" \
  "$freshness_root/fresh.txt" --max-age 3600 --min-bytes 1 \
  && printf 'Output is recent and nonempty\n'

python3 - <<'PY_REPAIR'
from pathlib import Path
Path("stale.txt").write_text("ready\n", encoding="utf-8")
PY_REPAIR
python3 check_freshness.py stale.txt --max-age 3600