Python中解析非标准格式的多重JSON对象流-Python教程-PHP中文网

Python中解析非标准格式的多重JSON对象流

php中文网

发布： 2025-12-06 17:11:02

原创

709人浏览过

Python中解析非标准格式的多重JSON对象流

本教程旨在解决接收到多个json对象以非标准格式（即没有外部数组括号和逗号分隔）直接连接的场景。我们将介绍一种python解析策略，通过识别json对象的结束和开始标记来精确分割数据流，从而实现对每个独立json对象的成功解析和处理。

理解非标准JSON数据流

在处理API响应或日志数据时，我们通常期望接收到符合RFC 8259标准的JSON格式，例如单个JSON对象或一个包含多个JSON对象的数组。然而，在某些情况下，可能会遇到一种非标准的数据流，其中多个独立的JSON对象直接连接在一起，没有外部的方括号 [] 包裹，也没有逗号 , 进行分隔。这种格式使得标准的 json.loads() 函数无法直接解析整个字符串，因为它不是一个有效的JSON文档。

以下是一个典型的非标准JSON数据流示例：

{
"self": "https://example1.com",
"key": "keyOne",
"name": "nameOne",
"emailAddress": "mailOne",
"avatarUrls": { /* ... */ },
"displayName": "displayNameOne",
"active": true,
"timeZone": "Europe",
"locale": "en_UK"
}
{
"self": "https://example2.com",
"key": "keyTwo",
"name": "nameTwo",
"emailAddress": "mailTwo",
"avatarUrls": { /* ... */ },
"displayName": "displayNameTwo",
"active": false,
"timeZone": "Europe",
"locale": "en_US"
}

登录后复制

在这种结构中，一个JSON对象的结束符 } 之后紧跟着下一个JSON对象的开始符 {，它们之间可能只有换行符。我们的目标是识别这些边界，并将数据流分割成独立的、可解析的JSON字符串。

解析策略：基于行识别边界

由于每个JSON对象都是一个独立的实体，其结束标志是 }，而下一个对象的开始标志是 {。我们可以利用这一特性，逐行扫描整个数据流。当检测到当前行是 { 并且前一行是 } 时，就意味着我们找到了两个JSON对象之间的边界。

立即学习“Python免费学习笔记（深入）”；

Freepik Mystic

Freepik Mystic 是一款革命性的AI图像生成器，可以直接生成全高清图像

174

查看详情

具体的解析步骤如下：

将整个非标准JSON数据流按行分割。
遍历这些行，维护一个当前正在构建的JSON对象的起始索引。
当发现一个行以 } 结束，并且紧接着的下一行以 { 开始时，这表明前一个完整的JSON对象已经结束。此时，将从起始索引到当前行（包含）的所有行拼接起来，形成一个完整的JSON字符串。
使用 json.loads() 函数解析这个字符串，并将其添加到结果列表中。
更新起始索引为当前行的下一行，继续处理剩余数据。
处理完所有行后，最后一个JSON对象需要单独处理。

Python实现示例

下面是使用Python实现上述解析策略的示例代码：

import json

# 示例非标准JSON数据流
data_stream = '''
{
"self": "https://example1.com",
"key": "keyOne",
"name": "nameOne",
"emailAddress": "mailOne",
"avatarUrls": {
  "48x48": "https://test.com/secure/useravatar?avatarId=1",
  "24x24": "https://test.com/secure/useravatar?size=small&avatarId=1",
  "16x16": "https://test.com/secure/useravatar?size=xsmall&avatarId=1",
  "32x32": "https://test.com/secure/useravatar?size=medium&avatarId=1"
},
"displayName": "displayNameOne",
"active": true,
"timeZone": "Europe",
"locale": "en_UK"
}
{
"self": "https://example2.com",
"key": "keyTwo",
"name": "nameTwo",
"emailAddress": "mailTwo",
"avatarUrls": {
  "48x48": "https://test.com/secure/useravatar?avatarId=2",
  "24x24": "https://test.com/secure/useravatar?size=small&avatarId=2",
  "16x16": "https://test.com/secure/useravatar?size=xsmall&avatarId=2",
  "32x32": "https://test.com/secure/useravatar?size=medium&avatarId=2"
},
"displayName": "displayNameTwo",
"active": false,
"timeZone": "Europe",
"locale": "en_US"
}
'''

def parse_concatenated_json(data_string: str) -> list:
    """
    解析包含多个直接连接的非标准JSON对象的字符串。

    Args:
        data_string: 包含非标准JSON对象流的字符串。

    Returns:
        一个包含所有解析出的JSON字典的列表。
    """
    json_objects = []
    lines = data_string.strip().splitlines() # 移除首尾空白并按行分割

    current_object_start_line = 0 # 记录当前JSON对象开始的行索引

    for i, line in enumerate(lines):
        # 检查当前行是否为"{"且前一行是否为"}"
        # 并且确保不是第一个JSON对象的开始（i > 0）
        if i > 0 and line.strip() == "{" and lines[i-1].strip() == "}":
            # 找到一个JSON对象的边界，解析前一个对象
            json_segment = "\n".join(lines[current_object_start_line:i])
            try:
                json_objects.append(json.loads(json_segment))
            except json.JSONDecodeError as e:
                print(f"解析JSON片段失败: {e}\n片段内容:\n{json_segment}")
                # 根据实际需求选择是跳过、记录错误还是中断
                pass
            # 更新下一个JSON对象的起始行
            current_object_start_line = i

    # 处理最后一个JSON对象
    if current_object_start_line < len(lines):
        json_segment = "\n".join(lines[current_object_start_line:])
        try:
            json_objects.append(json.loads(json_segment))
        except json.JSONDecodeError as e:
            print(f"解析最后一个JSON片段失败: {e}\n片段内容:\n{json_segment}")

    return json_objects

# 调用解析函数
parsed_data = parse_concatenated_json(data_stream)

# 打印解析结果以验证
print(f"成功解析 {len(parsed_data)} 个JSON对象:")
for obj in parsed_data:
    print(json.dumps(obj, indent=2, ensure_ascii=False)) # 格式化输出，方便阅读

登录后复制

代码解析：

data_stream.strip().splitlines(): 首先使用 strip() 移除整个字符串开头和结尾的空白字符（包括换行），然后使用 splitlines() 将字符串分割成一个行的列表。
current_object_start_line: 这个变量用于标记当前正在构建的JSON对象的起始行索引。
for i, line in enumerate(lines): 遍历每一行及其索引。
if i > 0 and line.strip() == "{" and lines[i-1].strip() == "}": 这是核心判断逻辑。它检查：
- i > 0: 确保不是数据流的起始，因为第一个JSON对象之前没有 }。
- line.strip() == "{": 当前行去除空白后是否为 {。
- lines[i-1].strip() == "}": 前一行去除空白后是否为 }。
- 如果这三个条件都满足，则表明我们已经跨越了一个完整的JSON对象的边界。
json_segment = "\n".join(lines[current_object_start_line:i]): 将从上一个起始行到当前行之前的所有行拼接起来，形成一个完整的JSON字符串。使用 "\n".join() 保持原始的换行符，这对于 json.loads 解析多行JSON字符串是必要的。
json.loads(json_segment): 使用Python内置的 json 模块解析这个片段。
current_object_start_line = i: 更新起始行索引为当前行的索引，为解析下一个JSON对象做准备。
处理最后一个JSON对象: 循环结束后，最后一个JSON对象仍然在 lines 列表中，从 current_object_start_line 到列表末尾。因此，需要一个额外的步骤来解析它。

注意事项与总结

数据格式依赖性: 这种方法高度依赖于每个JSON对象以 } 结尾，并且下一个对象以 { 开始的特定模式。如果数据流中存在其他分隔符或更复杂的结构（例如，JSON对象内部的字符串也包含 { 或 } 但不在行首或行尾），则此方法可能需要调整或失效。
错误处理: 在实际应用中，应加入更健壮的错误处理机制。例如，如果 json.loads() 失败，应该捕获 json.JSONDecodeError 异常，并决定是跳过该片段、记录错误信息，还是终止解析过程。示例代码中已加入了基本的错误捕获。
性能考量: 对于非常大的数据流，逐行读取和字符串拼接可能会消耗较多的内存和CPU。如果性能是关键因素，可以考虑使用更流式（streaming）的解析器，或者在拼接字符串时优化内存使用。
数据清洗: 在 splitlines() 之前使用 strip() 是一个好习惯，可以去除整个数据流开头和结尾可能存在的无关空白字符。同样，在比较行内容时使用 line.strip() 可以忽略行内的前导/尾随空白。

通过上述基于行识别边界的策略，我们可以有效地解析和处理那些不符合标准JSON数组格式，而是以非标准方式直接连接的多个JSON对象流。这种方法为处理来自不规范数据源的JSON数据提供了一个实用的解决方案。

以上就是Python中解析非标准格式的多重JSON对象流的详细内容，更多请关注php中文网其它相关文章！