数据处理：json / csv / dataclasses（Python 从精通到入门 · 16）

现实世界的程序八成时间在处理数据；这篇把 JSON、CSV 和 dataclasses 三件套串起来，最后做一个能跑的小项目。

你将学到

• json.dumps/loads 的参数陷阱：ensure_ascii、indent、default
• datetime 等非标准类型怎么序列化
• csv 的 DictReader / DictWriter，分隔符与编码
• dataclasses：@dataclass、field、default_factory、asdict、排序
• 用 jsonschema 或手写逻辑校验畸形数据
• 一个「读 CSV → 处理 → 存 JSON」的完整小项目

前置知识

需要文件读写和 JSON 基础，见 上一篇：正则表达式 与 14-文件与路径处理。

json：dumps / loads

dumps（dump string）把对象变成 JSON 字符串，loads 反过来。带文件用 dump / load。三个高频参数：

import json

# 基本往返：dumps 转字符串，loads 转回对象
obj = {"name": "Alice", "tags": ["dev", "py"]}
s = json.dumps(obj)
print(s)                            # 输出: {"name": "Alice", "tags": ["dev", "py"]}
print(json.loads(s)["tags"])        # 输出: ['dev', 'py']

data = {"城市": "上海", "人口": 2400}

# ensure_ascii=True（默认）会把非 ASCII 转义成 \uXXXX，中文看着难受
print(json.dumps(data))
# 输出: {"\u57ce\u5e02": "\u4e0a\u6d77", "\u4eba\u53e3": 2400}

# ✅ 想保留中文，关掉它
print(json.dumps(data, ensure_ascii=False))
# 输出: {"城市": "上海", "人口": 2400}

# indent 美化缩进，方便人看（也更大）；sort_keys 让 key 有序，便于 diff / 测试比对
print(json.dumps({"b": 1, "a": 2}, sort_keys=True))   # 输出: {"a": 2, "b": 1}

JSON 只认 dict / list / str / int / float / bool / None。别的类型（datetime、set、自定义对象）需要 default 回调告诉它怎么转：

import json
from datetime import datetime, date

def json_default(o):
    if isinstance(o, (datetime, date)):
        return o.isoformat()          # 转 ISO 8601 字符串
    if isinstance(o, set):
        return sorted(o)              # set 转 list
    raise TypeError(f"不可序列化: {type(o)}")

data = {"created": datetime(2026, 10, 7, 10, 30), "roles": {"admin", "user"}}
print(json.dumps(data, ensure_ascii=False, default=json_default))
# 输出: {"created": "2026-10-07T10:30:00", "roles": ["admin", "user"]}

csv：DictReader / DictWriter

CSV 的坑比 JSON 多：编码、分隔符、逗号与引号。用 csv 模块而不是自己 split(",")。

import csv

# 写：DictWriter 按字典 key 写，表头顺序用 fieldnames 控制
rows = [
    {"id": 1, "name": "Alice", "score": 95},
    {"id": 2, "name": "Bob", "score": 88},
]
with open("scores.csv", "w", encoding="utf-8", newline="") as f:  # ⚠️ newline=""
    writer = csv.DictWriter(f, fieldnames=["id", "name", "score"])
    writer.writeheader()               # 写表头
    writer.writerows(rows)

# 读：DictReader 把每行变成字典，表头自动成 key
with open("scores.csv", encoding="utf-8", newline="") as f:
    for row in csv.DictReader(f):
        print(row["name"], int(row["score"]))   # 输出: Alice 95 / Bob 88

newline="" 是必须的——不写会在 Windows 上产生空行（csv 自己要管换行）。

分隔符用 delimiter 指定，比如制表符分隔的 TSV：

with open("data.tsv", encoding="utf-8", newline="") as f:
    reader = csv.reader(f, delimiter="\t")
    for row in reader:
        print(row)

编码提醒：Excel 存的中文 CSV 常是 GBK（encoding="gbk"），或带 BOM（用 utf-8-sig 吃掉）；判断错了就 UnicodeDecodeError。

dataclasses：给数据一个类型

字典能装数据，但没有字段约束、没有补全、没有类型。@dataclass 用几行样板换来一个真正的类。

from dataclasses import dataclass, field, asdict

@dataclass
class Student:
    name: str
    score: int
    tags: list[str] = field(default_factory=list)   # 可变默认值必须用 factory

s = Student("Alice", 95, ["math"])
print(s)            # 输出: Student(name='Alice', score=95, tags=['math'])
print(s.name)       # 输出: Alice
print(asdict(s))    # 输出: {'name': 'Alice', 'score': 95, 'tags': ['math']}

为什么可变默认值要 default_factory？因为普通默认值在函数定义时就求值一次，会被所有实例共享：

# ❌ 所有实例共享同一个 list，改动互相污染
@dataclass
class Bad:
    items: list = []          # 危险！

# ✅ 每个实例各得一个新 list
@dataclass
class Good:
    items: list = field(default_factory=list)

排序与比较：

from dataclasses import dataclass

# order=True 自动生成 < <= > >=，按字段声明顺序比较
@dataclass(order=True)
class Point:
    x: int
    y: int

pts = [Point(3, 1), Point(1, 5), Point(1, 2)]
for p in sorted(pts):
    print((p.x, p.y))
# 输出: (1, 2) / (1, 5) / (3, 1)   —— 先比 x，再比 y

不想参与比较的字段用 field(compare=False)；不想出现在 repr 里用 field(repr=False)。从字典还原成对象时，用 Student(**{k: v for k, v in row.items() if k in Student.__dataclass_fields__}) 过滤掉多余字段。

校验畸形数据

外部数据永远不可信。轻量校验用手写（零依赖），或 jsonschema（pip install jsonschema）。

# 手写：简单场景够用
def validate_student(d):
    if isinstance(d, dict) and isinstance(d.get("name"), str) and d["name"] \
            and isinstance(d.get("score"), int) and 0 <= d["score"] <= 100:
        return True
    raise ValueError("字段不合法")

print(validate_student({"name": "Alice", "score": 95}))   # 输出: True

# jsonschema：声明式，规则清晰，适合复杂结构
from jsonschema import validate, ValidationError

schema = {
    "type": "object",
    "required": ["name", "score"],
    "properties": {
        "name": {"type": "string", "minLength": 1},
        "score": {"type": "integer", "minimum": 0, "maximum": 100},
    },
}
try:
    validate(instance={"name": "Alice", "score": 95}, schema=schema)
    print("OK")   # 输出: OK
except ValidationError as e:
    print("校验失败:", e.message)

原则：在数据进入系统边界时校验，别让畸形数据一路流到深处才炸。

小项目：读 CSV → 处理 → 存 JSON

需求：读学生成绩 CSV，算每人总分和等级，按总分降序输出 JSON。

# process_scores.py —— 完整可运行
import csv
import json
from dataclasses import dataclass, asdict, field

@dataclass
class Result:
    name: str
    total: int
    level: str
    subjects: list[str] = field(default_factory=list)

def load_rows(path):
    """读 CSV，跳过空行并把分数转 int；坏行跳过并告警。"""
    rows = []
    with open(path, encoding="utf-8-sig", newline="") as f:
        for i, row in enumerate(csv.DictReader(f), start=2):
            try:
                rows.append({"name": row["name"].strip(),
                             "subject": row["subject"].strip(),
                             "score": int(row["score"])})
            except (KeyError, ValueError) as e:
                print(f"[跳过] 第 {i} 行: {row!r} ({e})")
    return rows

def level_of(total):
    return "A" if total >= 180 else "B" if total >= 150 else "C" if total >= 100 else "D"

def aggregate(rows):
    """按姓名聚合总分与科目。"""
    acc = {}
    for r in rows:
        item = acc.setdefault(r["name"], {"total": 0, "subjects": []})
        item["total"] += r["score"]
        item["subjects"].append(r["subject"])
    return [
        Result(name, v["total"], level_of(v["total"]), v["subjects"])
        for name, v in acc.items()
    ]

def main():
    rows = load_rows("scores.csv")
    results = sorted(aggregate(rows), key=lambda r: r.total, reverse=True)
    payload = [asdict(r) for r in results]

    with open("result.json", "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)

    print(f"处理完成，共 {len(payload)} 人")
    for r in payload:
        print(r["name"], r["total"], r["level"])

if __name__ == "__main__":
    main()

输入 scores.csv：

name,subject,score
Alice,math,95
Alice,english,88
Bob,math,70
Bob,english,65

运行后打印 处理完成，共 2 人 / Alice 183 A / Bob 135 C，并生成 result.json（片段）：

[{"name": "Alice", "total": 183, "level": "A", "subjects": ["math", "english"]},
 {"name": "Bob", "total": 135, "level": "C", "subjects": ["math", "english"]}]

这个骨架可复用到绝大多数「读表 → 统计 → 导出」脚本：load_rows 负责脏数据、aggregate 负责业务、main 负责编排。

常见坑

# ❌ 默认 ensure_ascii 把中文转义
json.dumps({"城市": "上海"})                  # 输出: {"\u57ce\u5e02": "\u4e0a\u6d77"}

# ✅ 关掉它
json.dumps({"城市": "上海"}, ensure_ascii=False)   # 输出: {"城市": "上海"}

# ❌ 直接 dump datetime 报 TypeError；dataclass 用可变默认值会实例间互相污染
# ✅ 前者用 default 回调，后者用 default_factory
json.dumps({"t": datetime.now()}, default=str)     # 输出: {"t": "2026-10-07 ..."}

@dataclass
class A:
    items: list = []                               # ❌ 危险
@dataclass
class B:
    items: list = field(default_factory=list)      # ✅

# ❌ CSV 读时不开 newline=""，Windows 上每行多一个空行；外部 JSON 直接下标取值
with open("a.csv", encoding="utf-8") as f: ...      # ❌
name = data["user"]["name"]                         # ❌ KeyError 随时炸

# ✅ 加 newline=""；用安全的取值路径 + 校验
with open("a.csv", encoding="utf-8", newline="") as f: ...
name = data.get("user", {}).get("name")
if name is None:
    raise ValueError("缺少 user.name")

小结

• ensure_ascii=False 保留中文，indent 美化和调试；非标准类型用 default 转换。
• CSV 永远配 newline=""，中文注意 utf-8-sig / gbk；读用 DictReader，写用 DictWriter。
• @dataclass 把字典升级成有类型的对象；可变默认值必须 default_factory；order=True 一键获得排序。
• 边界处校验外部数据，jsonschema 声明式，手写逻辑零依赖。
• 三段式脚本骨架：解析脏数据 / 纯函数业务 / main 编排，最好测试。

延伸阅读

• 官方文档：json、csv、dataclasses
• 第三方：pydantic（更强大的数据校验，值得一学）、jsonschema
• 下一篇 网络请求与 Web 基础 教你把这些数据处理能力接到网络上

---

上一篇：正则表达式 · 下一篇：网络请求与 Web 基础