小菜鸟

java菜鸟号正在起航

0%

python JSON解析

发表于 更新于 分类于 阅读次数： Valine：
本文字数： 5.3k 阅读时长 ≈ 5 分钟

Python 解析 JSON 日志：从一行数据到一份报告

写后端的人，天天跟日志打交道。现在大部分应用的日志都是 JSON 格式——每条日志是一个 JSON 对象，方便机器解析和分析。

Python 自带的 json 模块就是干这个的：把 JSON 字符串转成字典，或者把字典转成 JSON 字符串。用法就几个函数，但真正写脚本分析日志的时候，有些细节还是得注意。

核心就两个函数：loads 和 dumps

函数 干啥的 啥时候用
json.loads(s) JSON 字符串 → Python 字典 读日志行的时候用
json.dumps(obj) Python 字典 → JSON 字符串 写结果文件的时候用

JSON 字符串转字典：

1
2
3
4
5
6
7
import json

line = '{"url": "/api/user", "timeSpent": 45, "status": 200}'
data = json.loads(line)

print(data["url"])        # /api/user
print(data["timeSpent"])  # 45

字典转 JSON 字符串：

1
2
3
data = {"name": "Bob", "age": 25, "city": "上海"}
json_str = json.dumps(data, ensure_ascii=False)
# {"name": "Bob", "age": 25, "city": "上海"}

ensure_ascii=False 这个参数经常被忽略——不加的话，中文会变成 \u4e0a\u6d77，虽然能解析，但人没法看。

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

假设你有一份 API 日志 req_resp.log，每行是一个 JSON 对象：

1
2
3
4
5
{"url": "/api/user", "timeSpent": 45, "status": 200, "method": "GET"}
{"url": "/api/order", "timeSpent": 230, "status": 200, "method": "POST"}
{"url": "/api/product", "timeSpent": 89, "status": 500, "method": "GET"}
{"url": "/api/report", "timeSpent": 310, "status": 200, "method": "GET"}
{"url": "/api/search", "timeSpent": 120, "status": 200, "method": "POST"}

需求很简单：找出耗时超过 150ms 的请求，打印出来。

完整实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#!/usr/bin/env python3
import json

LOG_PATH = "req_resp.log"

with open(LOG_PATH, "r", encoding="utf-8") as f:
    for line_num, line in enumerate(f, 1):
        line = line.strip()
        if not line:
            continue
        
        try:
            data = json.loads(line)
        except json.JSONDecodeError as e:
            print(f"第 {line_num} 行 JSON 解析失败: {e}")
            continue
        
        time_spent = data.get("timeSpent", 0)
        if time_spent > 150:
            print(f"[慢请求] {data.get('method')} {data.get('url')} 耗时 {time_spent}ms")

运行结果

1
2
[慢请求] POST /api/order 耗时 230ms
[慢请求] GET /api/report 耗时 310ms

这个脚本虽然短，但包含了几个关键点：

  • 逐行读取：文件大也不怕，不会一次性加载到内存
  • 跳过空行：日志文件末尾经常有空行
  • 捕获 JSONDecodeError：日志里可能有格式错误的一行，不能因为一条坏数据让整个脚本崩溃
  • 用 get() 而不是 []：万一某条日志缺了某个字段，get() 返回默认值，[] 直接报 KeyError

把结果写到文件里

光打印不够，通常需要把筛选结果存下来，方便后续分析或者发给别人。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import json

LOG_PATH = "req_resp.log"
OUTPUT_PATH = "slow_requests.json"

slow_requests = []
error_count = 0

with open(LOG_PATH, "r", encoding="utf-8") as f:
    for line_num, line in enumerate(f, 1):
        line = line.strip()
        if not line:
            continue
        
        try:
            data = json.loads(line)
        except json.JSONDecodeError:
            error_count += 1
            continue
        
        if data.get("timeSpent", 0) > 150:
            # 只保留需要的字段
            filtered = {
                "url": data.get("url"),
                "method": data.get("method"),
                "timeSpent": data.get("timeSpent"),
                "status": data.get("status")
            }
            slow_requests.append(filtered)

# 写入结果文件
with open(OUTPUT_PATH, "w", encoding="utf-8") as out_f:
    json.dump(slow_requests, out_f, ensure_ascii=False, indent=2)

print(f"慢请求数量: {len(slow_requests)}")
print(f"解析失败: {error_count} 行")
print(f"结果已保存到 {OUTPUT_PATH}")

indent=2 让输出的 JSON 有缩进，方便人看。如果文件很大，去掉 indent 能省空间。

处理嵌套 JSON

有些日志的字段是嵌套的，比如这样：

1
2
3
4
5
6
7
8
{
    "request": {
        "url": "/api/user",
        "method": "GET",
        "headers": {"User-Agent": "Mozilla/5.0"}
    },
    "response": {"status": 200, "timeSpent": 45}
}

访问嵌套字段用连续的 [] 或者 get()：

1
2
3
url = data["request"]["url"]
method = data["request"]["method"]
time_spent = data["response"]["timeSpent"]

如果某个中间字段可能不存在，用 get() 加上空字典兜底：

1
url = data.get("request", {}).get("url")

这种写法在数据格式不完全统一的时候很好用，一行代码就把嵌套访问和默认值都处理了。

几个经常遇到的问题

1. JSON 解析失败

1
json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes

最常见的几个原因：

  • 用了单引号而不是双引号（JSON 标准要求双引号）
  • 末尾多了逗号
  • 行里有非 JSON 的前缀，比如 [INFO] 2024-01-15 {"url": "/api"}

解决办法： 捕获异常，打印出问题行：

1
2
3
4
5
try:
    data = json.loads(line)
except json.JSONDecodeError:
    print(f"解析失败的行: {line[:100]}")
    continue

如果日志行有固定前缀，用正则把 JSON 部分提取出来再解析。

2. 中文变成 Unicode

不加 ensure_ascii=False 的时候：

1
json.dumps({"name": "张三"})   # '{"name": "\\u5f20\\u4e09"}'

加上：

1
json.dumps({"name": "张三"}, ensure_ascii=False)   # '{"name": "张三"}'

3. 字段类型不一致

有的日志里 timeSpent 是数字，有的可能是字符串 "230"。解析的时候统一转换：

1
2
3
4
5
6
7
8
9
10
def safe_get_int(data, key, default=0):
    value = data.get(key)
    if value is None:
        return default
    try:
        return int(value)
    except (ValueError, TypeError):
        return default

time_spent = safe_get_int(data, "timeSpent", 0)

文件读写：load 和 dump

如果整个文件就是一个 JSON 对象（不是每行一个），用 json.load() 和 json.dump()：

1
2
3
4
5
6
7
# 读取
with open("config.json", "r", encoding="utf-8") as f:
    config = json.load(f)

# 写入
with open("output.json", "w", encoding="utf-8") as f:
    json.dump(config, f, ensure_ascii=False, indent=2)

但日志分析通常用逐行处理（每行一个 JSON），因为文件可能很大，一次性加载会撑爆内存。

相关文章

欢迎关注我的其它发布渠道