Python subprocess.run 如何安全传递多行标准输入：input、text 与返回码验收

批处理脚本经常需要把一段多行文本交给外部程序：例如把筛选后的记录送进校验器，再根据退出码决定是否落库。最容易出错的写法是把文本拼进命令字符串，既难处理换行，也容易把输入误当成命令参数。Python 的 subprocess.run 可以把参数列表、标准输入和结果检查拆开处理，关键是把 input、text、capture_output 与 returncode 配成一条可验收的链路。

多行文本优先放在 input，命令使用参数列表而不是字符串拼接；文本模式用 text=True，业务失败需要抛异常时再打开 check=True，否则保留 returncode 做分支判断。

先把调用链拆成四个可检查节点

这类脚本的边界可以压缩成四个节点：run 创建并等待子进程，input 把多行字符串送入标准输入，communicate 负责完成输入输出交互，最后由 returncode 告诉调用方结果。把四者放在同一条路径上，排错时就不会只盯着命令文本。

第一步：用参数列表和 input 传递整段文本

下面用 Python 自带的 cat（Windows 可换成能从标准输入读取文本的等价程序）演示输入路径。命令本身只表达程序和参数，业务文本独立放在 input 中。

import subprocess

payload = "order_id= A-1001\nstatus=ready\n"
result = subprocess.run(
    ["cat"],
    input=payload,
    text=True,
    capture_output=True,
    check=False,
)

assert result.returncode == 0
assert result.stdout == payload
print(result.stdout)

capture_output=True 等价于同时把标准输出和标准错误接到管道，返回对象中的 stdout、stderr 会在子进程结束后可读。这里用 check=False 是为了让代码自己检查 returncode；如果任务失败还需要记录上下文，这种方式更容易组织日志。

第二步：明确 text、bytes 和编码的边界

当 text=True 时，input 应是字符串，输出也按文本读取。若不打开文本模式，默认的输入输出是字节，直接把字符串传给 input 会触发类型错误。需要固定编码时，显式写 encoding="utf-8"，不要把机器默认编码当成协议。

result = subprocess.run(
    ["cat"],
    input="中文行\n第二行\n",
    text=True,
    encoding="utf-8",
    capture_output=True,
)
if result.returncode != 0:
    raise RuntimeError(result.stderr.strip() or "外部校验器失败")

如果对方程序按二进制协议读取，才改成 input=b"..." 并按字节处理输出。不要同时传 input 和 stdin，官方接口会把这两个含义视为冲突。

第三步：决定是检查返回码还是直接抛异常

同一套调用可以有两种失败策略。脚本需要继续汇总多条记录时，保留 check=False 并读取 returncode；单个关键步骤失败就应该中断时，改成 check=True，非零退出会抛出 CalledProcessError。

try:
    subprocess.run(
        ["cat"],
        input="must-pass\n",
        text=True,
        capture_output=True,
        check=True,
    )
except subprocess.CalledProcessError as exc:
    detail = (exc.stderr or "").strip()
    raise RuntimeError(f"外部步骤失败，退出码={exc.returncode}，信息={detail}") from exc

不要只用“有没有异常”判断业务成功：程序可能正常启动但返回非零，也可能输出了看似正常的文本却没有完成预期动作。对于校验器、编译器、格式转换器这类工具，退出码和关键输出应一起验收。

第四步：给长时间任务加超时和可读诊断

外部程序卡住时，timeout 会让 run 终止并等待子进程结束，然后重新抛出 TimeoutExpired。生产脚本要把输入摘要、命令参数和超时秒数写入日志，但不要把整段可能含敏感数据的输入原样落盘。

try:
    result = subprocess.run(
        ["cat"],
        input=payload,
        text=True,
        capture_output=True,
        timeout=5,
    )
except subprocess.TimeoutExpired as exc:
    raise RuntimeError("外部校验器超过 5 秒未完成") from exc

超时不是“返回码失败”的同一种情况。监控上应分开统计，否则你会把程序异常慢和程序主动拒绝混成一类，后续无法判断该调资源、查死锁还是修业务输入。

常见误区与上线前检查

• 不要把用户输入直接拼成 shell=True 的命令字符串；参数列表能减少空格、换行和特殊字符带来的歧义。
• capture_output=True 适合输出可控的工具；输出可能无限增长时，应改用文件或更底层的流式读取方案。
• 固定文本协议时同时确认换行符和编码；在 Linux、Windows 与容器环境各跑一遍样例。
• 记录 returncode、超时和截断后的 stderr，但避免记录凭据、完整请求体或个人数据。

相关问题

input 能和 stdin=subprocess.PIPE 一起写吗？

不应一起传。使用 input 时，Python 会自动为标准输入建立管道；同时设置 stdin 会造成参数冲突。

为什么设置了 text=True 还要写 encoding？

text=True 只切换到文本语义，显式编码能避免不同机器的默认编码差异。跨平台处理中文或协议文件时建议明确写出。

什么时候应该使用 check=True？

当非零退出就意味着当前流程不能继续时使用。若要收集多个子任务结果或展示失败输出，则保留返回对象并按 returncode 分支更灵活。

把输入、输出和退出状态分开验收，脚本才真正知道外部程序做没做完。先从参数列表和 input 开始，再按任务需要补上编码、超时和异常策略，通常比修补一条复杂 shell 字符串更稳。