Linux下使用Python实时处理终端输出的JSON数据

你可以通过两种方式实现跳过文件存储，直接让Python实时处理命令行程序的JSON输出：

方式一：命令行管道传递输出 #

在终端中用管道将命令行程序的输出直接传给Python脚本，脚本从标准输入逐行读取并解析。

命令行执行： #

your_cli_program | python parse_json.py

对应的parse_json.py脚本： #

import sys
import json

# 逐行读取标准输入的内容
for line in sys.stdin:
    cleaned_line = line.strip()
    # 跳过空行
    if not cleaned_line:
        continue
    try:
        # 解析当前行的JSON数据
        json_data = json.loads(cleaned_line)
        # 提取你需要的字段，示例提取"id"和"result"
        extracted_data = {
            "id": json_data.get("id"),
            "result": json_data.get("result")
        }
        # 后续处理：比如打印到终端
        print(json.dumps(extracted_data))
        # 如果要保存到文件，可用追加模式写入
        # with open("filtered_output.jsonl", "a", encoding="utf-8") as f:
        #     f.write(json.dumps(extracted_data) + "\n")
    except json.JSONDecodeError:
        # 处理JSON解析失败的行，输出到标准错误流
        print(f"无效的JSON行: {cleaned_line}", file=sys.stderr)

方式二：Python脚本内部调用命令行程序 #

如果不想依赖终端管道，可直接在Python脚本中启动命令行程序，实时捕获并处理其输出。

示例脚本： #

import subprocess
import json

# 启动目标命令行程序，捕获标准输出
cli_process = subprocess.Popen(
    ["your_cli_program"],  # 替换为你的命令行程序路径及参数
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True  # 以文本模式读取输出，无需手动解码字节
)

# 实时读取程序输出的每一行
for line in cli_process.stdout:
    cleaned_line = line.strip()
    if not cleaned_line:
        continue
    try:
        json_data = json.loads(cleaned_line)
        # 提取指定字段，示例提取"timestamp"和"message"
        filtered = {
            "timestamp": json_data.get("timestamp"),
            "message": json_data.get("message")
        }
        # 这里执行你的后续逻辑，比如打印或存储
        print(filtered)
    except json.JSONDecodeError:
        print(f"解析JSON失败: {cleaned_line}", file=sys.stderr)

# 等待命令行程序执行完毕，处理错误输出
cli_process.wait()
if cli_process.returncode != 0:
    error_msg = cli_process.stderr.read()
    print(f"命令行程序执行出错: {error_msg}", file=sys.stderr)

关键注意点 #

• 确保你的命令行程序输出是每行一个独立的JSON对象（即JSON Lines格式），如果是跨多行的复杂JSON结构，需要调整读取逻辑（比如缓存内容直到完整JSON解析完成）。
• 异常处理很重要，避免因单条无效JSON导致整个脚本崩溃。
• 流式处理的方式不会将所有输出加载到内存，适合处理大量实时输出的场景。

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