Python subprocess如何处理持续输出的CSV流式数据？

你猜的没错，subprocess.run()确实会等待子进程完全结束才返回结果，没法处理持续流式输出的场景。用subprocess.Popen()是正确的思路，下面是具体实现步骤和代码：

核心思路 #

1. 用Popen启动子进程，将标准输出重定向到管道
2. 循环读取管道中的每一行输出（匹配目标脚本每隔一段时间输出一行CSV的逻辑）
3. 实时解析CSV行转为JSON格式
4. 处理进程退出、用户中断等异常情况

示例代码 #

import subprocess
import csv
import json
import sys
from io import StringIO

def main():
    # 构造调用applicationAPI.py的命令参数，替换为你的实际配置
    cmd = [
        sys.executable,
        "applicationAPI.py",
        "--target_ip", "192.168.1.100",
        "--username", "admin",
        "--password", "your_password",
        "-u", "60"
    ]

    # 启动子进程，开启行缓冲+文本模式，确保实时获取输出
    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        text=True,
        bufsize=1  # 行缓冲避免输出被系统缓存阻塞
    )

    csv_fields = None

    try:
        # 循环读取子进程输出的每一行
        for line in proc.stdout:
            line = line.strip()
            if not line:
                continue

            # 第一行是CSV表头，先记录字段
            if not csv_fields:
                csv_fields = line.split(',')
                continue

            # 解析数据行，转为字典后转JSON
            data_dict = dict(zip(csv_fields, line.split(',')))
            json_result = json.dumps(data_dict, indent=2)
            # 这里可以根据需求将JSON输出到其他系统/存储
            print("转换后的JSON数据：")
            print(json_result)

    except KeyboardInterrupt:
        # 用户按Ctrl+C时，终止子进程
        print("\n收到中断信号，正在终止子进程...")
        proc.terminate()
        proc.wait()
    except Exception as e:
        print(f"处理出错: {str(e)}")
        proc.terminate()
        proc.wait()
    finally:
        # 确保子进程被清理，避免僵尸进程
        if proc.poll() is None:
            proc.kill()

if __name__ == "__main__":
    main()

关键细节说明 #

• bufsize=1：开启行缓冲模式，避免子进程的输出因系统默认的块缓冲而无法实时读取（如果对方脚本已经手动调用stdout.flush()，这个参数可以省略，但加上兼容性更好）
• text=True：让subprocess直接返回字符串而非字节流，简化后续解析逻辑
• CSV转JSON：用表头字段和数据行直接映射成字典，再通过json.dumps()转换，逻辑简洁高效
• 异常处理：捕获KeyboardInterrupt确保子进程能被正常终止，避免残留进程

内容的提问来源于stack exchange，提问作者Meme-ento