Python脚本如何同时处理管道输入、执行子进程并记录原始数据？

核心问题是管道输入是一次性流式数据，读一次就会耗尽，所以必须先把原始数据完整保存到变量，再将变量作为输入传递给jq命令，同时保证jq的输出能正常流向后续管道。

修正后的脚本 #

import subprocess
import sys

def main():
    # 读取全部管道输入，保存原始数据
    raw_data = sys.stdin.read()
    if not raw_data:
        sys.exit(0)
    
    # 用列表形式定义命令，避免shell=True的安全风险与引号转义问题
    cmd = [
        "/usr/bin/jq",
        "-c",
        '.user |= if test("^[Bb]asic ") then .[6:] | @base64d | gsub(":.*$"; "") else . end'
    ]
    
    try:
        # 将原始数据作为输入传给jq，同时让jq的输出直接流入后续管道
        subprocess.run(
            cmd,
            input=raw_data.encode(),  # 转为字节流传递给子进程
            check=True,
            stdout=sys.stdout,
            stderr=sys.stderr
        )
    except subprocess.CalledProcessError:
        # 执行失败时将原始数据追加写入错误日志（用追加避免覆盖旧记录）
        with open('/path/to/parseerror.log', 'a') as pfile:
            pfile.write(raw_data + '\n')

if __name__ == "__main__":
    main()

关键说明 #

• 提前保存原始数据：用sys.stdin.read()一次性读取全部管道输入到内存，确保后续无论是传给jq还是写入日志都能复用这份数据。
• 显式传递输入给子进程：通过subprocess.run的input参数把原始数据传给jq，替代让jq直接读取已经被耗尽的标准输入。
• 避免shell=True：将命令拆分为列表形式，既规避了shell注入风险，也省去了引号转义的麻烦。
• 继承标准流：stdout=sys.stdout保证jq的处理结果能直接进入后续管道（比如rotatelogs），stderr=sys.stderr让jq的错误信息正常输出到控制台或指定日志。
• 日志追加模式：用'a'模式打开错误日志，防止每次错误覆盖历史记录，便于排查问题。

测试验证 #

执行你的测试命令：

echo '{"user": "Basic dGVzdDp0ZXN0"}' | python script.py | rotatelogs /var/log/test.log 86400

此时：

1. 脚本会完整保存原始JSON数据
2. jq处理后的结果会正常传递给rotatelogs
3. 若jq执行失败（比如输入不是合法JSON），原始数据会被追加到错误日志中

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