在Python中处理jq编译器异常并验证jq查询输入无关性

核心问题分析 #

原代码使用os.popen无法捕获jq的执行错误——它仅返回标准输出，无法获取命令退出状态或错误输出。而jq执行失败时（语法错误、执行时逻辑错误）会返回非0退出码，并将错误信息输出到标准错误流，这是我们捕获异常的关键依据。此外，用echo传递JSON字符串容易引发引号转义问题，更安全的方式是通过进程标准输入传递数据。

改进后的代码实现 #

import json
import os
import subprocess

def test_jq_filter(input_json, jq_filter):
    """测试jq过滤器，返回(是否执行成功, 输出/错误信息)"""
    try:
        # 调用jq，通过stdin传递JSON，捕获输出和错误
        result = subprocess.run(
            ["jq", jq_filter],
            input=json.dumps(input_json).encode('utf-8'),
            capture_output=True,
            text=True,
            check=False  # 不自动抛出异常，手动处理退出码
        )
        if result.returncode == 0:
            return (True, result.stdout.strip())
        else:
            return (False, result.stderr.strip())
    except Exception as e:
        return (False, str(e))

def process_sample_files(samples_dir="samples"):
    # 用于测试的通用JSON集合，覆盖对象、数组等不同结构
    generic_json_samples = [
        {"color": "red", "value": "#f00"},
        {
            "id": "0001",
            "type": "donut",
            "name": "Cake",
            "image": {"url": "images/0001.jpg", "width": 200, "height": 200},
            "thumbnail": {"url": "images/thumbnails/0001.jpg", "width": 32, "height": 32}
        },
        [100, 500, 300, 200, 400]
    ]
    
    compatible_files = []
    incompatible_files = []

    # 遍历目录下所有JSON样本文件
    for filename in os.listdir(samples_dir):
        if not filename.endswith(".json"):
            continue
        file_path = os.path.join(samples_dir, filename)
        print(f"处理文件: {filename}")

        try:
            with open(file_path, 'r') as f:
                sample_data = json.load(f)
            original_input = sample_data['input']
            jq_filter = sample_data['jq_solution']

            # 第一步：测试原输入是否能正常执行
            print("  测试原输入...")
            success_original, _ = test_jq_filter(original_input, jq_filter)
            if not success_original:
                print(f"  ❌ 原输入执行失败，过滤器本身存在错误")
                incompatible_files.append(filename)
                continue

            # 第二步：测试通用JSON样本，全部通过才算兼容
            print("  测试通用JSON兼容性...")
            all_pass = True
            for test_json in generic_json_samples:
                success_generic, err_msg = test_jq_filter(test_json, jq_filter)
                if not success_generic:
                    print(f"    ❌ 通用JSON测试失败: {err_msg}")
                    all_pass = False
                    break

            if all_pass:
                print("  ✅ 所有测试通过，过滤器兼容通用输入")
                compatible_files.append(filename)
            else:
                incompatible_files.append(filename)

        except KeyError as e:
            print(f"  ❌ 文件格式错误: 缺失必填键 {e}")
            incompatible_files.append(filename)
        except json.JSONDecodeError:
            print(f"  ❌ 文件不是有效的JSON格式")
            incompatible_files.append(filename)
        except Exception as e:
            print(f"  ❌ 处理出错: {str(e)}")
            incompatible_files.append(filename)
        print("-" * 50)

    # 输出统计结果
    print("\n=== 最终统计 ===")
    print(f"兼容通用输入的文件 ({len(compatible_files)}个):")
    for fname in compatible_files:
        print(f"- {fname}")
    print(f"\n不兼容的文件 ({len(incompatible_files)}个):")
    for fname in incompatible_files:
        print(f"- {fname}")

# 启动批量处理
process_sample_files()

关键改进说明 #

• 用subprocess替代os.popen：通过subprocess.run可以获取命令退出码、标准输出和错误输出，精准捕获jq的编译错误（比如语法错误）和执行错误（比如访问不存在的字段）。
• 安全传递JSON数据：通过进程标准输入传递JSON，避免了echo命令的引号转义问题，同时杜绝shell注入风险。
• 批量自动化处理：自动遍历指定目录下的所有JSON样本，无需手动输入文件编号。
• 严格兼容性判定：只有过滤器在原输入和所有通用JSON样本上都执行成功，才判定为不依赖特定输入结构。
• 清晰错误反馈：输出每个文件的测试细节和错误原因，方便定位问题。

兼容性判定依据 #

如果过滤器在通用JSON测试中出现以下错误，说明它依赖特定输入结构：

• Cannot index array with string "user"：在数组上访问对象字段（原输入是对象，但通用输入是数组）
• startswith is not defined：在非字符串类型上调用字符串方法
• Cannot iterate over null (null)：尝试遍历不存在的字段

而通用操作（比如. | length、map(.)、@base64等）则能在所有通用JSON样本上执行成功。

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