如何在Python中使用print向subprocess管道写入字符串？

核心问题是subprocess.Popen的stdin是二进制模式的_io.BufferedWriter，只能写入字节，但print需要文本模式的文件对象来处理字符串。我们需要一个中间层，自动把字符串编码成字节后再写入二进制流。

方案一：用codecs模块快速包装（推荐） #

这是最简便的方法，利用Python内置的codecs模块实现文本到二进制的转换：

import subprocess
import codecs

# 替换成你的实际命令
command = ["echo", "-n"]
popen = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)

# 用指定编码（比如utf-8）包装二进制流，得到文本模式的对象
text_stdin = codecs.getwriter('utf-8')(popen.stdin)

# 直接用print写入字符串
print("Hello, 测试文本", file=text_stdin)
# 手动刷新缓冲，确保内容立即传给子进程
text_stdin.flush()

# 关闭输入并等待子进程结束
popen.stdin.close()
output = popen.communicate()[0]
print("子进程输出:", output)

codecs.getwriter返回的对象实现了文本模式文件的接口，会自动把你传入的字符串编码成字节，再写入底层的二进制管道。

方案二：自己实现包装类（理解原理） #

如果想搞清楚底层逻辑，可以手动写一个简单的类，只需要实现write和flush方法就能满足print的需求：

import subprocess

class TextPipeWrapper:
    def __init__(self, binary_stream, encoding='utf-8'):
        self.binary_stream = binary_stream
        self.encoding = encoding

    def write(self, string):
        # 将字符串编码为字节，写入二进制流
        self.binary_stream.write(string.encode(self.encoding))

    def flush(self):
        # 转发刷新操作，确保内容不留在缓冲里
        self.binary_stream.flush()

# 使用示例
command = ["cat"]
popen = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)

text_stdin = TextPipeWrapper(popen.stdin)
# 把这个text_stdin传给只会print的代码即可
print("自定义包装类测试", file=text_stdin)
text_stdin.flush()

popen.stdin.close()
output = popen.communicate()[0]
print("子进程输出:", output)

注意事项 #

• 编码要和子进程预期的一致，比如子进程用GBK读取，这里就要把编码改成gbk
• 必须手动调用flush()，因为print默认有行缓冲，不刷新的话子进程可能收不到数据
• 操作完成后要关闭stdin并等待子进程结束，避免产生僵尸进程

内容的提问来源于stack exchange，提问作者user1806566