python JSON解析

Python JSON 解析详解：日志分析实战

JSON（JavaScript Object Notation）是一种轻量级的数据交换格式，在日志记录、API 交互等场景中广泛使用。Python 标准库 json 提供了简洁的 API 用于 JSON 数据的解析（字符串转字典）和序列化（字典转字符串）。本文将以日志分析为例，详细介绍 Python 处理 JSON 数据的方法。

JSON 解析核心方法

Python 的 json 模块提供了两个核心函数用于 JSON 处理：

方法	功能	适用场景
`json.loads(s)`	将 JSON 字符串 `s` 解析为 Python 字典 / 列表	处理内存中的 JSON 字符串（如日志行）
`json.dumps(obj)`	将 Python 字典 / 列表序列化为 JSON 字符串	将数据写入文件或通过网络传输
`json.load(f)`	从文件对象 `f` 中读取 JSON 数据并解析	直接读取 JSON 文件
`json.dump(obj, f)`	将 Python 对象序列化并写入文件 `f`	直接写入 JSON 文件

日志分析实战：筛选慢请求

以分析 JSON 格式的请求日志为例，假设日志文件 req_resp.log 中每行是一个 JSON 对象，包含 timeSpent（请求耗时，单位 ms）等字段。我们需要筛选出耗时超过 150ms 的请求。

完整代码实现

#!/usr/bin/python3
import json

# 日志文件路径
log_path = "/data/req_resp.log"

# 打开文件并逐行解析
with open(log_path, "r", encoding="utf-8") as f:
    for line_num, line in enumerate(f, 1):  # 枚举行号（从1开始）
        try:
            # 解析 JSON 字符串为字典
            log_data = json.loads(line.strip())  # strip() 去除换行符等空白
            
            # 检查是否包含 timeSpent 字段
            if "timeSpent" not in log_data:
                print(f"警告：第 {line_num} 行缺少 timeSpent 字段")
                continue
            
            # 获取耗时并筛选
            time_spent = log_data["timeSpent"]
            # 确保 timeSpent 是数字（避免类型错误）
            if isinstance(time_spent, (int, float)) and time_spent > 150:
                print(f"慢请求（耗时 {time_spent}ms）：{log_data}")
                
        except json.JSONDecodeError as e:
            # 处理 JSON 解析错误（如格式错误）
            print(f"解析错误（第 {line_num} 行）：{e}")
        except Exception as e:
            # 处理其他异常（如字段类型错误）
            print(f"处理错误（第 {line_num} 行）：{e}")

代码解析

（1）文件读取与逐行处理

使用 with open(...) 上下文管理器打开文件，自动处理资源释放（无需手动 close()）。
enumerate(f, 1) 用于获取行号（从 1 开始），便于定位错误。

（2）JSON 解析（`json.loads`）

line.strip() 去除每行首尾的空白字符（如换行符 \n），避免解析错误。
json.loads(line) 将 JSON 字符串转换为 Python 字典（dict），便于通过键名访问字段（如 log_data["timeSpent"]）。

（3）异常处理

json.JSONDecodeError：捕获 JSON 格式错误（如括号不匹配、逗号错误）。
检查 timeSpent 字段是否存在，避免 KeyError。
用 isinstance(time_spent, (int, float)) 确保字段类型为数字，避免比较时的 TypeError。

（4）筛选逻辑

仅当 timeSpent 大于 150ms 时，打印该请求的详细信息。

进阶技巧

提取特定字段

若只需关注部分字段（如 url、timeSpent、status），可针对性提取，减少输出冗余：

# 提取特定字段
if "timeSpent" in log_data and log_data["timeSpent"] > 150:
    filtered = {
        "url": log_data.get("url", "未知URL"),
        "timeSpent": log_data["timeSpent"],
        "status": log_data.get("status", "未知状态")
    }
    print(filtered)

使用 dict.get(key, default) 避免因字段不存在导致的 KeyError。

写入筛选结果到文件

将筛选出的慢请求写入新文件，便于后续分析：

# 打开输出文件
with open("slow_requests.log", "w", encoding="utf-8") as out_f:
    with open(log_path, "r", encoding="utf-8") as in_f:
        for line in in_f:
            try:
                log_data = json.loads(line)
                if log_data.get("timeSpent", 0) > 150:
                    # 序列化并写入（ensure_ascii=False 保留中文）
                    json.dump(log_data, out_f, ensure_ascii=False)
                    out_f.write("\n")  # 每行一个 JSON 对象
            except:
                continue

json.dump(..., ensure_ascii=False) 确保中文等非 ASCII 字符正常显示（而非 Unicode 编码）。

处理超大日志文件

对于 GB 级别的大日志，readlines() 会占用大量内存，建议逐行读取（如上述代码），或分块处理。

常见问题与解决方案

问题场景	解决方案
JSON 字符串包含中文	解析 / 序列化时指定 `encoding="utf-8"`，并在 `json.dump` 中加 `ensure_ascii=False`。
日志行格式不标准（非纯 JSON）	先用字符串处理（如正则匹配）提取 JSON 部分，再解析。
嵌套 JSON 结构	通过多级键访问（如 `log_data["request"]["headers"]["User-Agent"]`）。
字段类型不固定（如有时是字符串有时是数字）	解析后先判断类型（`isinstance()`），再转换（如 `int(time_spent_str)`）。