(二十一)把采集 CSV 变成可解释的数组
1. 保持字段含义再清洗记录
现在使用第九章导出的 title,finished,tags 三列。finished 是 0 或 1,tags 单元格保存 JSON 字符串数组;先读业务结构,再建立数值表示。
| 书名 | 已读 | 标签 |
|---|---|---|
| Python 入门 | 否 | 基础、语法 |
| NumPy 实践 | 是 | 数值 |
| Pygame 小游戏 | 否 | 游戏 |
| HTTP 与数据采集 | 否 | 网络 |
| 算法笔记 | 是 | 算法 |
| Python 项目实践 | 否 | 工程 |
自拟分析任务: 核对已读比例、每本标签数量、每个标签覆盖的书籍数,以及标签之间的共现次数。报告同时保留清洗说明,方便查回哪些记录没有参加统计。
# 项目文件:analysis_project/src/readingarray/records.py
# 先按 CSV 和 JSON 的结构读取,再把已经核对的字段交给数组计算。
import csv
import json
from dataclasses import dataclass
from pathlib import Path
@dataclass
class Book:
title: str
finished: bool
tags: tuple[str, ...]
def parse_book(row):
if None in row or any(value is None for value in row.values()):
raise ValueError("列数与表头不一致")
title = row["title"].strip()
if not title:
raise ValueError("书名为空")
flag = row["finished"].strip()
if flag not in {"0", "1"}:
raise ValueError("finished 不是 0 或 1")
raw_tags = json.loads(row["tags"] or "[]")
if not isinstance(raw_tags, list) or any(
not isinstance(tag, str) for tag in raw_tags
):
raise ValueError("tags 不是字符串数组")
# 去掉标签两侧空白、空标签及重复项,保留首次出现次序。
tags = tuple(dict.fromkeys(tag.strip() for tag in raw_tags if tag.strip()))
return Book(title, flag == "1", tags)空白先在各自字段内处理:书名两端空白去掉,空书名记录原因;状态必须读出明确的 0 或 1,缺失状态不能直接转成未读。标签先由 JSON 解释,再去掉空白、空标签和重复标签。
# 续写同一文件:analysis_project/src/readingarray/records.py
# 逐行校验 CSV,同一编号重复时保留首次记录,并把冲突行写入问题列表。
def load_books(path: Path):
books, issues, seen = [], [], {}
with path.open(encoding="utf-8-sig", newline="") as stream:
reader = csv.DictReader(stream, strict=True)
fields = reader.fieldnames
if (
fields is None
or len(fields) != 3
or set(fields) != {"title", "finished", "tags"}
):
raise ValueError("表头应为 title、finished、tags 三列")
for number, row in enumerate(reader, start=1):
try:
book = parse_book(row)
except ValueError as error:
issues.append(f"第 {number} 条记录:{error}")
continue
if book.title in seen:
reason = "重复记录" if book == seen[book.title] else "同名记录内容冲突"
issues.append(f"第 {number} 条记录:{reason},保留首次记录")
continue
seen[book.title] = book
books.append(book)
if not books:
raise ValueError("没有可分析的书籍记录;" + ";".join(issues))
return books, issues同名记录沿用前面书籍库的识别方式。完全重复的只留第一次;同名却状态或标签不同的,也留下冲突说明,报告中显示保留了首次记录。CSV 本身的引号结构损坏则让读取失败,先修复文件结构,再继续分析。
2. 每本书占一行,每个标签占一列
已读状态用一维 bool_ 数组保存。标签矩阵用 0、1 表示某本书是否带有该标签;词表按首次出现次序组织为“基础、语法、数值、游戏、网络、算法、工程”。
# 项目文件:analysis_project/src/readingarray/arrays.py
# 行始终对应同一本书;标签列按首次出现次序建立,矩阵元素表示有无该标签。
from dataclasses import dataclass
import numpy as np
@dataclass
class Catalog:
titles: tuple[str, ...]
vocabulary: tuple[str, ...]
finished: np.ndarray
tags: np.ndarray
def build_arrays(books):
if not books:
raise ValueError("需要至少一本书建立目录数组")
titles = tuple(book.title for book in books)
vocabulary = tuple(dict.fromkeys(tag for book in books for tag in book.tags))
columns = {tag: index for index, tag in enumerate(vocabulary)}
finished = np.array([book.finished for book in books], dtype=np.bool_)
tags = np.zeros((len(books), len(vocabulary)), dtype=np.int64)
for row, book in enumerate(books):
for tag in book.tags:
tags[row, columns[tag]] = 1
return Catalog(titles, vocabulary, finished, tags)建立 shape (6,7) 的零矩阵,遇到标签就把对应格设成 1。第一本书前两列为 1,后面五本各在对应列有一个 1。title、finished 和 tags 的行始终保持同一本书的对应关系。
这里使用循环把字符串标签定位到列,之后的整列统计和矩阵运算交给 NumPy。数值化先确定含义,再选择哪一部分适合批量处理。