Python 等待文件生成：超时轮询、原子交付与退出码

导出任务还在写 JSON，下游脚本已经开始读取：有时找不到文件，有时读到半截内容。把 sleep(10) 调长只能碰运气，还会让正常流程白等。

本文面向初学者到中级读者，用 Python 标准库实现带超时的文件等待，并让生产者通过原子替换交付完整结果。适合本机小型报表、批处理摘要和构建后的辅助任务，不需要安装包或修改调度器。

1. 方法：先约定交付，再轮询

每轮任务使用新的输出目录，避免把上轮结果当成新结果。
生产者写同目录临时文件，关闭后用 os.replace() 发布最终路径，之后不再修改它。
消费者只在路径缺失时等待；读到无效 JSON、目录或权限错误就立即失败。
用 time.monotonic() 计算剩余等待时间，睡眠不超过剩余预算。

Python 的 monotonic 文档说明，该时钟不受系统时钟更新影响，适合计算经过时长。os.replace 文档说明，成功的重命名操作在 POSIX 下是原子的，跨文件系统则可能失败。因此临时文件放在目标文件旁边。

这里的交付协议比“文件大小连续两次没变”可靠：写入进程可能暂停，大小稳定不能证明已完成。更完整的写入流程见原子保存 JSON 文件；本文重点是消费端的等待和失败边界。

2. 准备：使用独立演示目录

需要 Python 3.9 或更新版本，无第三方依赖。示例在 Linux、Python 3.11.4 下测试，命令适用于 bash 或 zsh。在同一个终端依次执行：

python3 --version
mkdir wait-file-demo
cd wait-file-demo
mkdir run-001

重复实验时换新的目录名，不要复用已有的 run-001/result.json。把下一节保存为当前目录的 wait_json.py。

3. 实现：只对缺失文件重试

"""Wait for a small, trusted local JSON output. Python 3.9+, no dependencies."""
import argparse
import json
import math
from pathlib import Path
import sys
import time


def seconds(value):
    number = float(value)
    if not math.isfinite(number) or not 0 <= number <= 86400:
        raise argparse.ArgumentTypeError("must be finite and between 0 and 86400")
    return number


def wait_json(path, timeout, interval, clock=time.monotonic, sleep=time.sleep):
    deadline = clock() + timeout
    first = True
    while first or clock() < deadline:
        first = False
        try:
            with path.open(encoding="utf-8") as stream:
                return json.load(stream)
        except FileNotFoundError:
            remaining = deadline - clock()
            if remaining <= 0:
                break
            sleep(min(interval, remaining))
    raise TimeoutError(f"no file before deadline: {path}")


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("file", type=Path)
    parser.add_argument("--timeout", type=seconds, default=30.0,
                        help="wait budget in seconds; 0 means one immediate attempt")
    parser.add_argument("--interval", type=seconds, default=0.2,
                        help="poll interval in seconds; minimum 0.001")
    args = parser.parse_args()
    if args.interval < 0.001:
        parser.error("--interval must be at least 0.001 seconds")
    try:
        data = wait_json(args.file, args.timeout, args.interval)
    except TimeoutError as exc:
        print(f"TIMEOUT: {exc}", file=sys.stderr)
        return 1
    except (OSError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2
    except KeyboardInterrupt:
        return 130
    print(json.dumps({"status": "ready", "data": data}, ensure_ascii=True))
    return 0


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

--timeout 0 仍会尝试读取一次，适合立即检查。其他情况下，首次尝试立即开始，后续只在截止时间之前开始；临近截止时用剩余时间限制睡眠。系统调度可能使实际睡眠更长，所以这不是实时保证。

这里只读取可信本地目录中的小型普通 JSON 文件，不要传入 FIFO、设备、攻击者可控路径或可能挂起的网络挂载。超时限制的是轮询等待，不能中断已经开始的文件打开、读取或 JSON 解析。json.load() 不会检查业务字段，相关行为见 JSON 官方文档。

4. 运行：模拟延迟交付

先查看帮助：

python3 wait_json.py --help

启动一个后台生产者，同时让消费者等待。命令需要在没有启用 set -e 的交互终端中执行，以便保存两个进程的退出码：

python3 - <<'PY_PRODUCER' &
import json
import os
from pathlib import Path
import tempfile
import time

output = Path("run-001/result.json")
pending = None
try:
    time.sleep(0.3)  # Simulate work before publishing.
    with tempfile.NamedTemporaryFile(
        mode="w", encoding="utf-8", dir=output.parent,
        prefix=".pending-", delete=False,
    ) as stream:
        pending = Path(stream.name)
        json.dump({"rows": 3}, stream, allow_nan=False)
        stream.write("\n")
    os.replace(pending, output)
finally:
    if pending is not None:
        pending.unlink(missing_ok=True)
PY_PRODUCER
producer_pid=$!
python3 wait_json.py run-001/result.json --timeout 5 --interval 0.1
consumer_exit=$?
wait "$producer_pid"
producer_exit=$?
printf 'consumer=%s producer=%s\n' "$consumer_exit" "$producer_exit"

NamedTemporaryFile()负责生成唯一临时文件名；delete=False 让关闭后的文件仍可重命名，finally 清理本次临时文件。正常输出为：

{"status": "ready", "data": {"rows": 3}}
consumer=0 producer=0

若手动执行太慢，消费者启动前结果可能已发布，立即成功同样正确。示例没有承诺断电持久性，也不解决多个生产者写同一路径的冲突；这里是一轮一个生产者、一个新目录。

退出码	含义
0	已读取并解析 JSON；标准输出为一行结果。
1	截止时仍未取得文件；标准错误以 `TIMEOUT:` 开头。
2	内容、访问或命令行参数错误；只写标准错误。
130	等待或读取期间收到 Ctrl-C。

接入下一步时用 python3 wait_json.py ... && 下一步命令 的条件执行方式，不要用 || true 隐藏失败。若要检查长期保留的产物是否过期，改用产物修改时间与空文件检查，但修改时间也不能代替本轮任务标识。

5. 验证：成功、超时和错误退出

下面只创建和清理自己的测试文件，允许子进程返回非零，然后检查实际返回码：

python3 - <<'PY_CHECK'
from pathlib import Path
import subprocess
import sys
import tempfile

with tempfile.TemporaryDirectory(prefix="checks-", dir=".") as folder:
    root = Path(folder)
    (root / "good.json").write_text('{"rows": 3}\n', encoding="utf-8")
    (root / "bad.json").write_text('{"rows":', encoding="utf-8")
    cases = [
        ("ready", "good.json", "0", 0, "--interval", "0.05"),
        ("missing", "missing.json", "0.1", 1, "--interval", "0.05"),
        ("invalid", "bad.json", "0", 2, "--interval", "0.05"),
        ("bad-argument", "good.json", "0", 2, "--interval", "0"),
    ]
    for label, name, timeout, expected, flag, value in cases:
        result = subprocess.run(
            [sys.executable, "wait_json.py", str(root / name),
             "--timeout", timeout, flag, value],
            capture_output=True, text=True, timeout=5,
        )
        assert result.returncode == expected, (label, result.stderr)
        if expected == 0:
            assert '"status": "ready"' in result.stdout
            assert not result.stderr
        else:
            assert not result.stdout and result.stderr
        print(f"PASS {label} exit={result.returncode}")
PY_CHECK

预期输出：

PASS ready exit=0
PASS missing exit=1
PASS invalid exit=2
PASS bad-argument exit=2

延迟交付由第 4 节演示覆盖。想把这种等待扩展成定期监控，可参考固定间隔任务与 monotonic，但不要把无限循环当成超时策略。

6. 排错：不要靠延长 sleep 掩盖问题

现象	原因与直接处理
一直 `TIMEOUT:`	确认两端路径和当前目录一致，检查上面保存的 `producer_exit`。使用绝对路径定位问题后，再根据正常耗时设置 `--timeout`。
`ERROR: Expecting ...`	最终文件是残缺 JSON。让生产者先关闭临时文件再替换；不要把解析错误当成缺失文件无限重试。
总是立即成功，却是旧数据	复用了上轮路径。给每轮任务创建新的输出目录；不要在共享目录中盲目删除旧文件。
`Permission denied` 或路径是目录	改为当前用户可读的普通文件路径，并检查父目录访问权限；不要直接以 root 重试。
超时后进程还没返回	正在阻塞的 I/O 不受这个轮询预算控制。排查挂载或文件类型，必要时把整个消费者放进有外部时限的受管子进程。

7. 总结

先规定完整文件如何交付，再用单调时钟和有限重试等待它。只重试缺失路径，解析和权限错误及时失败，每轮使用新目录避免误读旧结果。这个实现适合本机小型文件协作；需要跨机器通知、强制 I/O 时限或多生产者协调时，应换用对应的任务管理机制。