subprocess输出读取失败时，如何回退并捕获原始输出？

问题背景 #

使用subprocess执行命令时，若遇到UnicodeDecodeError，需要查看原始输出复现问题，但无法每次重新运行命令获取输出。尝试手动控制解码逻辑（不使用text/universal_newlines参数），但遇到seek方法无效的问题，导致无法保存解码失败的原始内容。

尝试的代码 #

import io
import logging
import pprint
import shutil
import subprocess
import tempfile


def run_cmd(cmd):
    ps = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=None, text=False)

    data = None

    # Decode with a wrapper instead of text to force utf-8 instead of local
    # encoding (cp1252 on windows).
    output = io.TextIOWrapper(ps.stdout, encoding="utf-8")
    try:
        data = output.read()
    except UnicodeDecodeError:
        with tempfile.NamedTemporaryFile(delete=False) as f:
            logging.warning("UnicodeDecodeError.")
            output.seek(0)
            shutil.copyfileobj(output, f)
            logging.error("UnicodeDecodeError. Wrote input to file: %s", f.name)
        raise

    return ps.wait(), data


pprint.pprint(run_cmd(["dir"]))  # My command actually contacts a server and sometimes fails.

遇到的问题 #

• 解码失败时，调用output.seek(0)后复制内容到临时文件，但文件始终为空
• 具体错误：UnicodeDecodeError: 'utf-8' codec can't decode byte 0x97 in position 337: invalid start byte
• 疑问：为何seek在此场景下不生效？（根据文档，不支持seek应抛出错误）

原因分析 #

TextIOWrapper包装的是subprocess的stdout管道对象，管道是不可回溯的流式对象：

1. 管道数据一旦被读取（哪怕是解码过程中预读的部分字节），就无法再从管道中重新获取
2. TextIOWrapper在解码时会预读字节缓冲区，即使解码失败，已经读取的字节也不会放回管道，因此seek(0)无法回到原始数据的开头，后续读取自然为空

通用解决方法 #

方法1：先读取所有原始字节，再尝试解码 #

先将subprocess的stdout全部读取为字节串，保存原始数据后再进行解码。即使解码失败，原始字节已留存，可用于排查问题。

import logging
import pprint
import subprocess
import tempfile


def run_cmd(cmd):
    ps = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=None, text=False)
    # 先读取所有原始字节
    raw_bytes = ps.stdout.read()
    returncode = ps.wait()
    
    data = None
    try:
        data = raw_bytes.decode("utf-8")
    except UnicodeDecodeError:
        with tempfile.NamedTemporaryFile(delete=False, mode='wb') as f:
            f.write(raw_bytes)
            logging.error("UnicodeDecodeError. Wrote raw bytes to file: %s", f.name)
        raise
    
    return returncode, data


pprint.pprint(run_cmd(["dir"]))

方法2：逐块读取并记录原始数据 #

如果命令输出极大，无法一次性加载到内存，可逐块读取字节，同时记录原始数据块。遇到解码失败时，拼接已读取的原始块保存即可。

import logging
import pprint
import subprocess
import tempfile


def run_cmd(cmd):
    ps = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=None, text=False)
    raw_chunks = []
    data_chunks = []
    
    try:
        while True:
            chunk = ps.stdout.read(4096)
            if not chunk:
                break
            raw_chunks.append(chunk)
            # 逐块解码，一旦失败立即触发异常
            data_chunks.append(chunk.decode("utf-8"))
        data = ''.join(data_chunks)
    except UnicodeDecodeError:
        raw_bytes = b''.join(raw_chunks)
        with tempfile.NamedTemporaryFile(delete=False, mode='wb') as f:
            f.write(raw_bytes)
            logging.error("UnicodeDecodeError. Wrote raw bytes to file: %s", f.name)
        raise
    finally:
        returncode = ps.wait()
    
    return returncode, data


pprint.pprint(run_cmd(["dir"]))

补充说明 #

• 方法1实现简单，适合大多数场景；若输出过大可能占用较多内存
• 方法2内存占用可控，适合大输出场景，但逻辑稍复杂
• 避免用TextIOWrapper包装管道流处理此类需求，管道流不支持随机访问，且TextIOWrapper的预读机制会导致数据丢失

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