用 Python 执行外部命令：超时、错误处理与日志文件

Python 流水线调用外部工具，工具已经报错，后面的步骤却继续读取不完整的结果。另一种情况是命令一直等待输入，而任务在后台运行，根本没有人会回答。终端输出一旦丢失，排查就更麻烦。

这篇教程写一个小型命令执行函数：遇到错误就停止，限制等待时间，并把输出写入新的日志文件。演示直接用 Python 作为子进程，不需要额外安装工具，就能验证每种情况。

1. 明确执行函数的职责

这个函数每次运行一个可信的可执行程序。参数用列表传入，标准输入关闭，标准输出和标准错误一起写入磁盘。每次运行都要使用新的日志文件名。它不自动重试，也不管理任务队列。

关键行为见 Python 的 subprocess.run 文档：check=True 会在非零退出时抛出异常；达到 timeout 后，会终止并等待直接子进程，再抛出超时异常。进程创建阶段可能延迟超时。使用参数列表可以处理带空格的路径，不必手动拼接 shell 引号。

2. 下载并运行成功示例

需要 Python 3.10 或更新版本，无第三方依赖。本文示例在 macOS、Python 3.14.7 下验证。下面的 shell 命令适用于 bash、zsh 等 POSIX shell。

将 run_logged.py 下载到一个新的工作目录，然后运行：

python3 --version
python3 run_logged.py --help
mkdir subprocess-demo
python3 run_logged.py ok subprocess-demo/ok.log
cat subprocess-demo/ok.log

最后两条命令的预期输出：

OK: log=subprocess-demo/ok.log
processed 3 records

重复实验时换一个新路径。示例会主动保留已经存在的日志。

3. 完整实现

"""Run a trusted command with a fresh log and a timeout. Python 3.10+."""
import argparse
import math
from pathlib import Path
import subprocess
import sys


def run_logged(argv, log_path, *, timeout=60):
    """Keep stdout/stderr on disk; propagate launch, exit and timeout errors."""
    if isinstance(argv, (str, bytes)) or not argv:
        raise ValueError("argv must be a nonempty argument sequence")
    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError("timeout must be a finite positive number")
    # Exclusive creation prevents accidental replacement of an earlier log.
    with Path(log_path).open("xb") as log:
        return subprocess.run(
            argv,
            stdin=subprocess.DEVNULL,
            stdout=log,
            stderr=subprocess.STDOUT,
            shell=False,
            check=True,
            timeout=timeout,
        )


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("mode", choices=["ok", "fail", "timeout"])
    parser.add_argument("log", type=Path, help="new log path; parent must exist")
    args = parser.parse_args()
    programs = {
        "ok": "print('processed 3 records', flush=True)",
        "fail": "import sys; print('invalid input', file=sys.stderr, flush=True); sys.exit(2)",
        "timeout": "import time; print('started', flush=True); time.sleep(30)",
    }
    command = [sys.executable, "-u", "-c", programs[args.mode]]
    try:
        run_logged(command, args.log, timeout=1 if args.mode == "timeout" else 10)
    except subprocess.CalledProcessError as exc:
        print(f"FAILED: exit={exc.returncode}; log={args.log}", file=sys.stderr)
        return 1
    except subprocess.TimeoutExpired:
        print(f"TIMEOUT: log={args.log}", file=sys.stderr)
        return 1
    except (OSError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(f"OK: log={args.log}")
    return 0


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

演示中的 -c 程序是我们自己编写的固定字符串。sys.executable 选择运行父脚本的 Python 解释器。接入自己的流水线时，把这个命令列表换成实际可执行程序及其参数即可。

日志以 xb 打开，路径已存在时会在启动命令前报错。二进制模式保留子进程输出的原始字节，父目录必须提前创建。如果日志打开后程序启动失败，可能留下一个空日志；仅凭日志文件存在，不能判断任务已经运行。

4. 验证失败与超时

逐条运行下面的命令。两次调用执行脚本都会按设计返回退出状态 1，因此不要把这组演示放入遇错立即退出的脚本。

python3 run_logged.py fail subprocess-demo/fail.log
cat subprocess-demo/fail.log
python3 run_logged.py timeout subprocess-demo/timeout.log
cat subprocess-demo/timeout.log

失败示例输出：

FAILED: exit=2; log=subprocess-demo/fail.log
invalid input

超时示例输出：

TIMEOUT: log=subprocess-demo/timeout.log
started

exit=2 是子进程的退出状态；演示 CLI 将失败和超时统一映射为自己的退出状态 1。直接导入 run_logged 时，原始异常会继续向上传递，方便业务代码分别处理。

超时示例让 Python 使用无缓冲输出，并主动刷新第一条消息。其他程序可能缓冲输出，被终止时日志会不完整，甚至为空。写入文件也意味着输出会持续消耗磁盘空间；长期任务还需要日志保留和大小策略。

5. 传入带空格的文件名

在临时工作目录中，将下面的代码保存为 count_demo.py，与 run_logged.py 放在一起。它创建一个小型工作脚本和输入文件，然后统计非空行。这两个演示用文件如果已经存在，会被覆盖。

import sys
from pathlib import Path
from run_logged import run_logged

worker = Path("count_records.py")
worker.write_text(
    "import sys\n"
    "from pathlib import Path\n"
    "with Path(sys.argv[1]).open(encoding='utf-8') as handle:\n"
    "    print(sum(1 for line in handle if line.strip()))\n",
    encoding="utf-8",
)
Path("sample records.txt").write_text("alpha\nbeta\ngamma\n", encoding="utf-8")
run_logged(
    [sys.executable, str(worker.resolve()), "sample records.txt"],
    "count.log",
    timeout=10,
)
assert Path("count.log").read_text().strip() == "3"
print("PASS: counted 3 nonempty lines")

运行 python3 count_demo.py，预期输出为 PASS: counted 3 nonempty lines。文件名作为列表中的一个元素传入；如果在字符串内再加引号，引号就会成为参数的一部分。

实际流水线应先验证结果，再记录完成状态。更新小型进度文件时，可以配合原子写入 JSON的方法；需要追踪实际输入数据时，可以保留一份输入文件校验清单。

6. 验证旧日志不会丢失

完成计数示例后，将下面的代码保存为 check_log.py，在同一目录运行 python3 check_log.py：

from pathlib import Path
from run_logged import run_logged
import sys

path = Path("count.log")
before = path.read_bytes()
try:
    run_logged([sys.executable, "-c", "print('replacement')"], path)
except FileExistsError:
    assert path.read_bytes() == before
    print("PASS: existing log preserved")
else:
    raise AssertionError("Expected FileExistsError")

这项检查确认重复运行不会悄悄清空上一次的日志。实际任务可以每次新建运行目录，或使用唯一的日志文件名，不要默认自动删除旧日志。

7. 排错与适用边界

现象	处理方法
`FileExistsError`	换一个尚不存在的日志路径。
`FileNotFoundError`	同时检查可执行程序与日志父目录。
`FAILED: exit=...`	阅读日志，查阅工具对退出状态的定义。
`TIMEOUT`	检查已写入的日志与任务量，再设定合适的时间限制。
工具立即提示缺少输入	这里关闭了 stdin；通过文件参数提供输入，或修改函数以支持 stdin。
日志乱码	使用子程序的输出编码打开文件；日志保留的是原始字节。

这个函数不会限制日志大小、回滚不完整的结果文件，也不保证终止直接子进程所启动的后代进程。会启动多个工作进程的工具，需要按操作系统设计进程树清理方式，或交给任务管理器。退出状态为零后，仍应检查业务输出。

命令和选项应来自可信来源。shell=False 不是沙箱，工具本身仍可能把某个参数解释成选项。Windows 批处理文件还有额外注意事项，见 Python 的 subprocess 安全说明。本文示例面向普通可执行程序和 Python 脚本。

8. 小结

为每次命令执行分配新的日志，明确超时时间，并分别处理错误。先验证失败场景，再把函数接入批量任务。保存进度前，分别检查进程是否成功、结果是否有效。