如何用Python等语言从含非JSON文本的文件中提取JSON对象

针对你需要处理1万+节点的Ansible facts导出文件（含非JSON文本+两个合法JSON对象）的场景，下面提供两种无需外部工具（如sed）、直接用Python解析提取的高效方法：

方法1：逐行扫描+括号计数法（内存友好，适合大文件） #

利用JSON对象的{}嵌套特征，逐行扫描文件，通过计数大括号的平衡状态来识别完整的JSON对象，自动跳过非JSON内容。

import json

def parse_facts_file(file_path):
    json_objects = []
    current_chunk = []
    brace_counter = 0

    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            line_content = line.strip()
            # 跳过空行和非JSON起始的无关行（未进入JSON收集状态时）
            if not line_content and brace_counter == 0:
                continue

            # 更新大括号计数
            brace_counter += line_content.count('{')
            brace_counter -= line_content.count('}')

            current_chunk.append(line_content)

            # 当括号平衡且已收集内容时，尝试解析JSON
            if brace_counter == 0 and current_chunk:
                try:
                    json_str = ' '.join(current_chunk)
                    json_obj = json.loads(json_str)
                    json_objects.append(json_obj)
                except json.JSONDecodeError:
                    # 解析失败则丢弃当前片段，继续扫描
                    pass
                finally:
                    current_chunk = []

    # 按需求组装自定义字典，比如区分系统facts和自定义facts
    return {
        "system_facts": json_objects[0],
        "custom_facts": json_objects[1]
    } if len(json_objects) >= 2 else {}

优势 #

• 逐行处理，内存占用极低，即使是大体积的facts文件也能轻松处理
• 自动忽略所有非JSON文本，无需提前预处理
• 解析逻辑稳定，能处理任意嵌套深度的JSON结构

方法2：正则匹配法（代码简洁，适合格式化JSON） #

Ansible导出的facts是格式化后的JSON，结构规整，可通过正则表达式匹配完整的嵌套JSON对象。

import json
import re

# 匹配完整嵌套JSON对象的正则（支持多层{}嵌套）
JSON_MATCHER = re.compile(r'\{(?:[^{}]|(?R))*\}', re.DOTALL)

def extract_facts_with_regex(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        full_content = f.read()

    # 提取所有匹配的JSON片段
    json_snippets = JSON_MATCHER.findall(full_content)
    valid_json = []
    for snippet in json_snippets:
        try:
            valid_json.append(json.loads(snippet))
        except json.JSONDecodeError:
            continue

    # 组装目标字典
    return {
        "system_facts": valid_json[0],
        "custom_facts": valid_json[1]
    } if len(valid_json) >= 2 else {}

优势 #

• 代码简洁，实现成本低
• 一次性读取文件后批量匹配，小文件场景下速度更快

批量处理优化建议 #

1. 并行处理：使用concurrent.futures.ProcessPoolExecutor开启多进程，利用多核CPU同时处理多个文件，大幅提升1万+节点的处理效率
2. 错误日志：添加日志模块记录解析失败的文件路径，方便后续排查异常节点
3. 资源控制：并行处理时限制进程数（建议等于CPU核心数），避免系统资源耗尽

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