春江暮客

春江暮客的个人学习分享网站

Check Scheduled Output Freshness with Python: File Age, Empty Files, and Exit Codes

2026-10-16 Technology
Check Scheduled Output Freshness with Python: File Age, Empty Files, and Exit Codes

A scheduled job logs “started,” and report.json exists in the output directory, but that file may be yesterday’s result. Checking only a process exit code or file existence can miss a job that never produces a new output.

This beginner-to-intermediate tutorial uses the Python standard library to check an output’s modification time, byte size, and file type, emit JSON, and stop downstream commands with an exit code. Use it for local reports, exports, and an additional check after a static-site build. No packages or scheduler changes are needed.

1. Method: define an acceptable output

We use a small set of rules:

  1. The path must provide metadata and point to a regular file.
  2. It must contain at least --min-bytes bytes, defaulting to 1; empty files trigger an alert.
  3. Its last modification must be no more than --max-age seconds ago; the exact limit is accepted.
  4. A modification time more than 5 seconds ahead of the current time also triggers an alert, so a future timestamp cannot keep old output looking healthy.

Python’s stat-result documentation defines st_mtime as the last content-modification time and st_size as the size in bytes. On Linux, st_ctime records metadata changes; it is not a creation timestamp. We use stat.S_ISREG() to identify regular files.

File existence and timely production are separate checks. If a scheduled script looks in the wrong directory, first follow scheduled-job path and environment troubleshooting to fix its working directory and interpreter path.

2. Setup and test files

Use Python 3.9 or newer, with no third-party dependencies. These examples were tested on Linux with Python 3.11.4. Commands work in bash or zsh.

  1. Create a fresh demo directory and keep subsequent commands in the same terminal. Choose a new directory name when repeating the experiment:
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
  1. Save the next section as check_freshness.py in the current directory. fresh.txt is recent, stale.txt is dated two hours ago, empty.txt is empty, and future.txt is dated one hour ahead.

os.utime() sets access and modification times. This changes only demo files, without waiting an hour or adjusting the system clock. Complete sections 4 and 5 within an hour of creating the fixtures; otherwise rerun the PY_DATA snippet above.

3. Complete implementation: inspect metadata and emit JSON

"""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())

The script reads metadata without reading or modifying the output’s contents. Path.stat() follows symbolic links and checks the target file; a broken link raises an error. Directories, devices, and FIFOs cannot qualify as valid outputs.

File timestamps and time.time() use comparable time references. Do not subtract a file timestamp from time.monotonic(). Wall-clock adjustments affect the age calculation; this example tolerates at most 5 seconds into the future. age_seconds is rounded to three decimal places for display; comparisons use the unrounded value.

4. Usage and exit codes

  1. Show help, then check the recent and old files:
python3 check_freshness.py --help
python3 check_freshness.py fresh.txt --max-age 3600
python3 check_freshness.py stale.txt --max-age 3600

The recent file produces output like this; age_seconds depends on when you run it:

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

The old file reports status: "alert" and problems: ["too_old"], with exit code 1. Empty files and future timestamps produce too_small and future_mtime, respectively. One file can have multiple problems.

  1. Use these exit-code rules in the caller:
Exit code Meaning
0 A regular file passes both age and size checks.
1 Output alert, missing file, access error, or incorrect file type prevents a successful check.
2 Invalid command-line arguments, such as --max-age 0 or a negative value.

When regular-file metadata is available, standard output contains one JSON line, including alert results. Missing paths, access errors, and file-type errors produce only ERROR: on standard error. argparse writes argument errors to standard error; see its official documentation. An absent successful JSON report does not mean there are no problems.

5. Validate failure branches without waiting

Run this self-check to verify both results and exit codes. It explicitly permits subprocess failures, then checks their actual return values:

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

Expected output:

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

The check does not rewrite the fixtures. Missing paths and directories must return nonzero, and invalid arguments must not silently succeed. Handle expected failures with conditionals when inspecting them in a shell; an unhandled exit code 1 can stop a script using set -e.

6. Integrate with an existing workflow and troubleshoot

  1. Check the actual result with absolute paths. This example still checks only the demo file and changes no scheduler settings:
python3 "$freshness_root/check_freshness.py" \
  "$freshness_root/fresh.txt" --max-age 3600 --min-bytes 1 \
  && printf 'Output is recent and nonempty\n'

The command after && runs only if the check succeeds, and a failed check retains its nonzero exit code. To detect a producer that never starts, invoke the checker periodically from an independent existing monitoring entry point. A checker embedded inside the producer cannot run when the producer never starts. For an hourly report, --max-age 5400 is a starting point that allows extra runtime and scheduling delay; adjust it to observed behavior.

After a Hugo build, you can also check the target HTML in that build’s output directory. Its timestamp cannot prove content or links are correct; continue with Hugo internal-link and image checks for content validation.

  1. Fix the cause indicated by each problem, then check again:
Symptom Concrete action
too_old Check the producer’s latest actual success, input, and output paths, then regenerate the output. If a copy tool preserves old modification times, monitor a status file generated after this run succeeds instead.
too_small Inspect producer errors and free disk space. Use --min-bytes 0 only when empty output is valid for the task. A byte threshold is not a record count.
future_mtime Check clocks on the producer and monitoring hosts, plus timestamps preserved during copying. Do not hide the problem by increasing limits indefinitely.
No such file Verify the mount, output name, and producing step with an absolute path.
Permission denied Check parent-directory access as the account running the checker. Being able to stat a file does not prove its contents are readable.
not a regular file Point to the specific file; a directory’s timestamp cannot stand in for a report’s timestamp.
  1. Regenerate the old demo output and verify recovery:
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

It should now return status: "ok" and exit code 0. In production, do not use touch merely to clear an alert: it does not generate missing business data.

A recent, nonempty file can still contain incomplete or incorrect data. The producer should validate content and replace the final file only after success; see atomic JSON writes in Python for a saving pattern. Metadata observed during a write cannot prove completion. Copies preserving modification times, filesystem timestamp precision, and network-filesystem caching can also affect the result.

7. Summary

Checking both maximum age and minimum size catches leftover old files and empty outputs while giving automation a clear exit code. Validate failure branches with controlled timestamps before checking production paths. Keep content validation and replacement after success in the producer so recent updates are not mistaken for correct results.

标签

1024 ablang adsense agents.md ai ai-agent ai-agents ai-seo algorithm amp antibodies apparmor argparse automation background backup bash batch-processing binarycif bioinformatics biopython blockchain boltz bootstrapping boxes bubblewrap c-index cca cdn checkpoint cli cloudflare codex cofoldarena copy cpu monitoring cron csv cuda curl data-leakage data-processing data-quality data-validation datascience datavisualization datetime debugging decimal deployment desktop-app devtools disk-space disown dns docker dovecot download economics electron encoding esm esm2 esm3 esmc esmfold2 fable faceswap fasta fastmcp faulthandler ffmpeg file-descriptors file-io flashppi flask folium frontend game generator git github-actions google google-research grep gzip harness hls html http hugo imports indexnow javascript jev json jsonl just k-means kaggle langfuse leecode linux list litellm llm llms.txt logging logs lollipop lsof m3u8 machine-learning macos manacher matplotlib mcp mirror mmcif model-evaluation monitoring mp3 mp4 mpnn multiomics mutation mysql nanobert nanobodies networkx nginx normalize numpy ollama omegatherm pandas password pathlib pdb pep-723 phaser pillow pip postfix preprocessing print probability protein-alignment protein-design protein-embeddings protein-interactions protein-language-models protein-stability protein-structure proxy pydantic pyecharts pyqt python python3 r raincloud reproducibility requests reservoir-sampling rfoptimization rg ripgrep rosettafold3 roundcube rrsi rsi rsync s-tui sampling scale scikit-learn scrapy screen seaborn security seo sequence-identity sha256 shell signals sklearn socket solana somaticsignatures spl sqlite ssh standardize static-site subprocess symlink sysadmin system-one tarfile tcp tensorflow tkinter tls toml trade tron tronpy troubleshooting turtle typesafe-ai ubuntu unicode usdt utf-8 uv vhhbert vite wallet webp wordcloud wordpress workflow yaml zipfile zoneinfo 迅雷解析

友情链接

其它