跳到正文
NKUGeek Hub
Geek Lab

实验 02 / 04 · 50 分钟

CSV 读取与数据汇总

把逐行解析组合成可定位错误、可重复运行的数据管线。

学习目标

  • 使用标准库正确处理 CSV、引号与 UTF-8 BOM。
  • 让错误包含具体行号，避免静默丢弃坏数据。
  • 按项目汇总时间，输出顺序稳定的报告。

前置知识

  • 完成实验 01，理解 Session 和 ValueError。
  • 能遍历列表并更新字典。
回顾实验 01：数据建模与输入校验

环境配置与运行

Python 3.11+ · 仅标准库 · 在本机终端运行

下载并解压课程包，打开其中的 python-engineering 目录。以下命令均在包含 lab01.py 的目录执行。首次学习时创建环境，后续实验复用同一个 .venv。

首次配置
py -3 --version
py -3 -m venv .venv
.\.venv\Scripts\python.exe --version
运行实验
.\.venv\Scripts\python.exe -X utf8 lab02.py
运行测试
.\.venv\Scripts\python.exe -X utf8 -m unittest test_lab02 -v
预期结果
{
  "session_count": 4,
  "total_minutes": 150,
  "projects": {
    "git": 30,
    "python": 90,
    "web": 30
  }
}

实验步骤

  1. 01

    检查输入契约

    打开 sessions.csv，确认表头严格等于 date,project,minutes。先手算 python 的两次 45 分钟，以及整体 150 分钟。

  2. 02

    拆开两个职责

    阅读 load_sessions 的文件上下文与错误行号，再阅读 summarize 的字典累计。先验证数据再计算汇总，坏行出现时整次处理失败。

  3. 03

    验证错误路径

    复制一份 CSV，把第三行分钟改成 -5。临时把脚本 sample 指向副本并运行，确认错误指出 line 3。随后还原代码并执行测试。

让 CSV 库负责解析格式

字段可能被引号包围，也可能含有逗号；split(',') 无法正确处理。DictReader 负责解析，parse_row 负责业务校验。utf-8-sig 同时接受普通 UTF-8 和带 BOM 的文件。

稳定输出便于比较

summarize 不修改输入列表，并按项目名排序。相同记录即使顺序不同，汇总结果也相同。重复记录代表多次学习，会重复累计；空数据必须保留表头。

完整代码

以下为课程包中的实际源码。后续实验复用前序模块，首次运行请下载完整课程包。

lab02.py
"""Lab 02: load CSV and build a deterministic summary."""

import csv
import json
from pathlib import Path

from lab01 import Session, parse_row


def load_sessions(path: Path) -> list[Session]:
    sessions = []
    with path.open(encoding="utf-8-sig", newline="") as source:
        reader = csv.DictReader(source, strict=True)
        if reader.fieldnames != ["date", "project", "minutes"]:
            raise ValueError("header must be: date,project,minutes")
        try:
            for row in reader:
                if None in row:
                    raise ValueError("unexpected extra columns")
                sessions.append(parse_row(row))
        except (ValueError, csv.Error) as error:
            raise ValueError(f"line {reader.line_num}: {error}") from None
    return sessions


def summarize(sessions: list[Session]) -> dict:
    totals: dict[str, int] = {}
    for session in sessions:
        totals[session.project] = totals.get(session.project, 0) + session.minutes
    return {
        "session_count": len(sessions),
        "total_minutes": sum(totals.values()),
        "projects": dict(sorted(totals.items())),
    }


if __name__ == "__main__":
    sample = Path(__file__).with_name("sessions.csv")
    print(json.dumps(summarize(load_sessions(sample)), indent=2))
sessions.csv
date,project,minutes
2026-09-28,python,45
2026-09-28,git,30
2026-09-29,python,45
2026-09-29,web,30

测试用例

执行上面的测试命令，应看到 8 个测试通过和 OK。先理解断言，再尝试引入错误验证测试有效。

test_lab02.py
test_lab02.py
import tempfile
import unittest
from pathlib import Path

from lab02 import load_sessions, summarize


class CsvTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.path = Path(self.temp.name) / "sessions.csv"

    def write(self, content, encoding="utf-8"):
        self.path.write_text(content, encoding=encoding)

    def test_sample_summary(self):
        sample = Path(__file__).with_name("sessions.csv")
        self.assertEqual(summarize(load_sessions(sample)), {
            "session_count": 4,
            "total_minutes": 150,
            "projects": {"git": 30, "python": 90, "web": 30},
        })

    def test_header_only(self):
        self.write("date,project,minutes\n")
        self.assertEqual(summarize(load_sessions(self.path)), {
            "session_count": 0, "total_minutes": 0, "projects": {},
        })

    def test_utf8_bom_and_quoted_comma(self):
        self.write('date,project,minutes\n2026-09-28,"docs, tests",20\n', "utf-8-sig")
        self.assertEqual(load_sessions(self.path)[0].project, "docs, tests")

    def test_wrong_header(self):
        self.write("day,project,minutes\n")
        with self.assertRaisesRegex(ValueError, "header"):
            load_sessions(self.path)

    def test_error_includes_line(self):
        self.write("date,project,minutes\n2026-09-28,python,-5\n")
        with self.assertRaisesRegex(ValueError, "line 2: minutes"):
            load_sessions(self.path)

    def test_extra_column(self):
        self.write("date,project,minutes\n2026-09-28,python,10,extra\n")
        with self.assertRaisesRegex(ValueError, "extra columns"):
            load_sessions(self.path)

    def test_missing_cell(self):
        self.write("date,project,minutes\n2026-09-28,python\n")
        with self.assertRaisesRegex(ValueError, "required fields"):
            load_sessions(self.path)

    def test_unclosed_quote(self):
        self.write('date,project,minutes\n2026-09-28,"python,10\n')
        with self.assertRaisesRegex(ValueError, "line"):
            load_sessions(self.path)


if __name__ == "__main__":
    unittest.main()

常见错误

header must be: date,project,minutes

表头区分大小写、顺序和空格。使用课程样例的原始表头，分隔符为英文逗号。

UnicodeDecodeError

将文件另存为 UTF-8 CSV。utf-8-sig 不负责将 GBK 或其他编码自动转换成 UTF-8。

line 2: unexpected extra columns

检查是否多了分隔符；项目名本身包含逗号时，需要用双引号包住整个字段。

验收清单

自评记录仅保存在当前浏览器，不代表自动测试结果或 Issue 认领状态。

从实验到贡献

增加按日期汇总函数

这是完成基线后的扩展任务。先查看 Issue 当前状态和留言，与维护者确认认领范围。

  • 新增 summarize_by_day，按 YYYY-MM-DD 升序返回每日分钟数。
  • 覆盖同一天多条记录、跨天和空列表。
  • 保留现有 summarize 的输出格式，补充样例与测试。
查看并认领 Issue #2
Python · CSV