Python subprocess.Popen管道与输出重定向至文件失效问题

你的代码不生效的核心问题有两个：

• 使用shell=True时，命令应该是完整字符串而非列表形式，否则管道|和重定向>会被当作普通参数传递给grep，而非由shell解析处理。
• 写入gzip文件时用了文本模式'w'，但gzip输出是二进制数据，必须用二进制模式'wb'。

下面提供两种可行的实现方式：

方法一：使用shell=True（简单直接） #

直接将完整的bash命令作为字符串传入，让shell处理管道和重定向逻辑：

import subprocess

# 完整命令字符串，shell=True会调用bash解析执行
command = "grep 'diff' test | gzip > test2.gz"
# 使用subprocess.run替代Popen，更简洁且默认等待命令完成
subprocess.run(command, shell=True, check=True)

• check=True会在命令执行失败时抛出异常，方便排查错误。
• 注意：如果命令中包含用户输入内容，不建议用这种方式，存在shell注入风险。

方法二：手动处理管道（更安全） #

不依赖shell，直接通过Python连接两个子进程的管道，同时以二进制模式写入文件：

import subprocess

# 执行grep命令，将输出导向管道
grep_process = subprocess.Popen(
    ['grep', 'diff', 'test'],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT
)

# 以二进制模式打开目标文件，将gzip的输出写入其中
with open('test2.gz', 'wb') as output_file:
    # 执行gzip命令，输入来自grep的管道输出
    gzip_process = subprocess.Popen(
        ['gzip'],
        stdin=grep_process.stdout,
        stdout=output_file,
        stderr=subprocess.STDOUT
    )
    # 关闭grep的stdout，避免子进程死锁
    grep_process.stdout.close()
    # 等待gzip命令执行完成
    gzip_process.wait()
    # 检查grep命令的执行状态
    if grep_process.returncode != 0:
        print(f"grep命令执行失败，返回码：{grep_process.returncode}")

这种方式完全由Python控制进程间的数据流，避免了shell注入风险，适合处理包含用户输入的场景。

内容的提问来源于stack exchange，提问作者R-addict