Python CSV 文件解析练习:从读取、清洗到汇总

2026-09-13 29 预计阅读时间: 1 分钟
来源: realpython.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:8 分钟

CSV 看起来只是用逗号分隔的文本,但真实数据通常还会包含表头、引号、空值、重复记录和格式不一致的字段。掌握 Python 标准库中的 csv 模块,不仅能完成日常数据处理,也能为面试中的文件解析题建立一套稳定的思路。

这篇文章用一组循序渐进的练习串起 CSV 处理流程:先读取记录,再进行字段转换和过滤,最后完成分组统计。示例数据是为了演示而构造的,你可以替换成自己的 CSV 文件继续练习。

先明确解析边界

处理 CSV 时,不建议直接使用 line.split(",")。字段本身可能包含逗号,例如:

id,name,comment
1,Alice,"likes Python, SQL, and data analysis"

如果直接按逗号切分,comment 会被错误拆成多个字段。Python 的 csv 模块会处理引号、转义字符和不同的换行方式,因此更适合作为默认方案。

下面的练习文件可以直接保存为 orders.csv

order_id,customer,amount,status
1001,Alice,120.50,paid
1002,Bob,89.00,cancelled
1003,Alice,45.50,paid
1004,Carol,,paid
1005,Bob,210.00,paid

练习一:读取并检查记录

使用 csv.DictReader 可以把每一行转换成字典。相比通过下标访问列,字典方式更容易读懂,也不容易因为列顺序变化而读错数据。

import csv
from pathlib import Path


def read_orders(path: str) -> list[dict[str, str]]:
    with Path(path).open("r", encoding="utf-8", newline="") as file:
        reader = csv.DictReader(file)
        if reader.fieldnames != ["order_id", "customer", "amount", "status"]:
            raise ValueError(f"unexpected columns: {reader.fieldnames}")

        return list(reader)


if __name__ == "__main__":
    orders = read_orders("orders.csv")
    for order in orders:
        print(order)

运行:

python parse_orders.py

这里有三个值得保留的习惯:使用 encoding="utf-8" 明确字符集,使用 newline="" 配合 csv 模块处理换行,并在输入边界检查表头。文件格式一旦变化,尽早失败通常比生成一份看似正常但实际错误的结果更容易排查。

练习二:转换类型并过滤数据

DictReader 返回的值都是字符串。金额需要转换成数值,缺失金额则必须明确处理。对于金额这类财务字段,生产代码通常应使用 decimal.Decimal,避免二进制浮点数带来的精度问题。

下面的程序只统计已支付且金额完整的订单:

import csv
from decimal import Decimal, InvalidOperation
from pathlib import Path


def paid_orders(path: str) -> list[dict[str, object]]:
    result = []

    with Path(path).open("r", encoding="utf-8", newline="") as file:
        for line_number, row in enumerate(csv.DictReader(file), start=2):
            raw_amount = (row.get("amount") or "").strip()
            if not raw_amount:
                print(f"skip line {line_number}: missing amount")
                continue

            try:
                amount = Decimal(raw_amount)
            except InvalidOperation:
                print(f"skip line {line_number}: invalid amount {raw_amount!r}")
                continue

            if row.get("status", "").strip().lower() != "paid":
                continue

            result.append({
                "order_id": row["order_id"].strip(),
                "customer": row["customer"].strip(),
                "amount": amount,
            })

    return result


if __name__ == "__main__":
    for order in paid_orders("orders.csv"):
        print(order["order_id"], order["customer"], order["amount"])

这道练习的重点不是语法,而是定义输入异常的行为:缺失金额是跳过、记为零,还是直接终止?不同业务有不同答案。面试中应把这个选择说出来,而不是让异常行为隐藏在类型转换代码里。

练习三:按客户汇总金额

读取、清洗和业务计算可以拆成独立步骤。这样既方便测试,也能在未来加入排序、输出新 CSV 或替换输入来源。

from collections import defaultdict
from decimal import Decimal


def total_by_customer(orders: list[dict[str, object]]) -> dict[str, Decimal]:
    totals: dict[str, Decimal] = defaultdict(Decimal)

    for order in orders:
        customer = str(order["customer"])
        amount = order["amount"]
        if not isinstance(amount, Decimal):
            raise TypeError("amount must be Decimal")
        totals[customer] += amount

    return dict(totals)

可以把前面的函数组合起来运行:

from parse_orders import paid_orders
from report import total_by_customer

orders = paid_orders("orders.csv")
for customer, total in sorted(total_by_customer(orders).items()):
    print(f"{customer}: {total:.2f}")

对于示例数据,Alice 的已支付金额应为 166.00Bob 的已支付金额应为 210.00Carol 因为金额为空,会在清洗阶段被跳过。

面试中可以继续追问

完成基础实现后,可以给自己增加约束,让练习更接近实际面试:

  • CSV 文件很大,不能一次性使用 list(reader),如何改成逐行处理?
  • 文件使用分号而不是逗号作为分隔符,应该修改哪里?
  • 某个字段包含换行符或逗号时,当前实现是否仍然正确?
  • 如何把汇总结果写入新的 CSV 文件?
  • 如何报告所有坏行,而不是遇到第一条错误就停止?
  • 如何为缺少表头、重复订单号和负金额增加校验?

大文件场景下,核心变化通常只是避免把所有记录放进内存:

import csv


def count_paid_rows(path: str) -> int:
    count = 0
    with open(path, "r", encoding="utf-8", newline="") as file:
        for row in csv.DictReader(file):
            if row.get("status", "").strip().lower() == "paid":
                count += 1
    return count

一份可执行的检查清单

练习 CSV 解析题时,可以按下面的顺序组织答案:

  1. 明确文件编码、分隔符和表头结构。
  2. 使用 csv.readercsv.DictReader,不要手写逗号切分。
  3. 为必填字段做存在性和格式校验。
  4. 在边界处完成字符串到 intDecimal 等类型的转换。
  5. 明确坏数据的处理策略,并保留行号或错误信息。
  6. 让读取、清洗、计算和写出保持相对独立。
  7. 用空文件、缺失字段、带逗号的字段和非法金额补充测试。

CSV 解析题真正考查的是对输入边界的判断。只会读出一行数据不难;能在数据不完美、文件很大、字段规则变化时保持结果可解释,才是可以迁移到真实项目中的能力。


相关推荐