English

Python 数据处理实战：清洗 CSV 与 JSON

AI 辅助生成本文由 AI 起草并经人工核查，示例均标明软件版本；发现错误请以官方文档为准。

Python 3.13

把前面学到的东西串起来：读一份不干净的 CSV，清洗、统计，输出一份 JSON 报告。脚本可直接运行、可测试。

环境准备

Python 3.13，只用标准库。

python -m venv .venv && source .venv/bin/activate

场景与数据

data/sales.csv（故意留了脏数据）：

date,city,amount,note
2026-10-01,上海,120.50,
2026-10-01,北京,
2026-10-02,上海,abc,金额异常
2026-10-02,,80.00,
2026-10-03,北京,200.00,

问题：空金额、非数字金额、空城市、首尾空格。

第一步：定义数据结构

用 dataclass 表达一行清洗后的数据，比到处用字典清楚：

# clean.py
from dataclasses import dataclass, asdict
from datetime import date
from pathlib import Path

@dataclass
class Sale:
    date: date
    city: str
    amount: float

第二步：逐行清洗

import csv
from collections import Counter

def parse_row(row: dict[str, str]) -> tuple[Sale | None, str | None]:
    """清洗一行。返回 (数据, 跳过原因)，跳过原因为 None 表示成功。"""
    city = (row.get("city") or "").strip()
    raw_amount = (row.get("amount") or "").strip()

    if not city:
        return None, "城市为空"
    if not raw_amount:
        return None, "金额为空"

    try:
        amount = float(raw_amount)
    except ValueError:
        return None, f"金额不是数字：{raw_amount!r}"

    if amount < 0:
        return None, f"金额为负：{amount}"

    return Sale(date.fromisoformat(row["date"].strip()), city, amount), None

要点：

  • 字段一律 strip()，空值用显式判断。
  • 转换失败用 try/except ValueError 而不是 if x.isdigit()（isdigit 认不了 120.50 和负数）。
  • 返回「(结果, 原因)」让上层能统计被丢弃的原因，而不是静默丢数据。

第三步：读取并统计

def load_sales(path: Path) -> tuple[list[Sale], Counter[str]]:
    sales: list[Sale] = []
    skipped: Counter[str] = Counter()

    with path.open(encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            sale, reason = parse_row(row)
            if sale is None:
                skipped[reason or "未知原因"] += 1
                continue
            sales.append(sale)

    return sales, skipped

第四步：聚合与报告

def summarize(sales: list[Sale]) -> dict:
    by_city: dict[str, float] = {}
    by_city_count: dict[str, int] = {}

    for sale in sales:
        by_city[sale.city] = by_city.get(sale.city, 0.0) + sale.amount
        by_city_count[sale.city] = by_city_count.get(sale.city, 0) + 1

    top_city = max(by_city, key=by_city.get) if by_city else None

    return {
        "total_records": len(sales),
        "total_amount": round(sum(s.amount for s in sales), 2),
        "average_amount": round(sum(s.amount for s in sales) / len(sales), 2) if sales else 0,
        "by_city": {city: round(amount, 2) for city, amount in sorted(by_city.items())},
        "records_by_city": dict(sorted(by_city_count.items())),
        "top_city": top_city,
    }

第五步：输出 JSON 报告

import json

def write_report(report: dict, out_path: Path) -> None:
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(
        json.dumps(report, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

完整入口

def main() -> int:
    base = Path(__file__).resolve().parent
    src = base / "data" / "sales.csv"
    out = base / "output" / "report.json"

    if not src.exists():
        print(f"找不到数据文件：{src}", file=sys.stderr)
        return 1

    sales, skipped = load_sales(src)
    report = summarize(sales)
    report["skipped"] = dict(skipped)

    write_report(report, out)
    print(f"有效 {len(sales)} 条，跳过 {sum(skipped.values())} 条 → {out}")
    for reason, count in skipped.most_common():
        print(f"  跳过原因 {reason}：{count} 条")
    return 0

if __name__ == "__main__":
    sys.exit(main())

运行：

python clean.py
# 有效 2 条，跳过 3 条 → .../output/report.json
#   跳过原因 金额为空：1 条
#   跳过原因 金额不是数字：'abc'：1 条
#   跳过原因 城市为空：1 条

报告内容：

{
  "total_records": 2,
  "total_amount": 320.5,
  "average_amount": 160.25,
  "by_city": { "北京": 200.0, "上海": 120.5 },
  "records_by_city": { "北京": 1, "上海": 1 },
  "top_city": "北京",
  "skipped": { "金额为空": 1, "金额不是数字：'abc'": 1, "城市为空": 1 }
}

让它可测试

parse_row 是纯函数，测试很容易写：

# tests/test_clean.py
from clean import parse_row

def test_parse_normal():
    sale, reason = parse_row({"date": "2026-10-01", "city": " 上海 ", "amount": "120.5"})
    assert reason is None
    assert sale.city == "上海"
    assert sale.amount == 120.5

def test_parse_bad_amount():
    sale, reason = parse_row({"date": "2026-10-02", "city": "上海", "amount": "abc"})
    assert sale is None
    assert reason is not None

load_sales 需要文件，用 pytest 的 tmp_path fixture 造临时 CSV（见 pytest 那一篇）。

常见错误与注意事项

1. 写 CSV 忘了 newline=""

Windows 下每行之间多出空行。

2. 用 float 做金额计算

0.1 + 0.2 != 0.3。金额精确计算用 decimal.Decimal：

from decimal import Decimal
Decimal("120.50") + Decimal("0.25")     # Decimal('120.75')

展示用 float 也无妨，但累加、对账必须用 Decimal。

3. 静默丢掉异常行

不统计跳过原因，最后对不上数也查不出问题。

4. 编码没写死

open(..., encoding="utf-8") 与 write_text(..., encoding="utf-8") 都要显式写。

5. 一个函数做了全部事

把「解析 / 统计 / 输出」拆开，才能单独测解析逻辑。

6. 用 isdigit() 判断数字

认不了小数与负数。直接 float() + try/except 最可靠。

小结

  • 清洗流程：读 → 逐行解析（记录跳过原因）→ 聚合 → 输出。
  • 用 dataclass 建模、Counter 统计跳过原因、pathlib 管路径。
  • 输出 JSON 用 ensure_ascii=False + indent=2。
  • 把解析逻辑抽成纯函数，配 pytest 做回归测试。

下一步

标签