流式处理 CSV 行

功能说明

stream_rows 用 csv.DictReader 逐行读取 CSV，并在同一个循环里把不合格的行写入拒绝文件，因此内存中始终只有当前行。函数先检查表头是否包含 id、email、age，再校验年龄范围与邮箱格式；合格行转换成统一类型后通过 yield 交给调用方，不合格行记录行号和原因。

关键代码

reader = csv.DictReader(source)
writer = csv.DictWriter(rejects, fieldnames=["line", "reason"])
writer.writeheader()
for line_no, row in enumerate(reader, start=2):
    try:
        age = int(row.get("age") or "")
    except ValueError:
        writer.writerow({"line": line_no, "reason": "age is not an integer"})
        continue
    yield {"id": row["id"], "email": email, "age": age}

两个文件都以 newline="" 打开，这是 csv 模块在 Windows 上正确处理换行的必要条件。

使用注意

• enumerate(..., start=2) 把表头算作第 1 行，拒绝文件里的行号与原始文件一致。
• 拒绝文件与源文件同时打开，生成器未消费完时两个句柄都会保持打开。
• 表头缺失会直接抛出 ValueError，不会产出任何数据。
• 示例把邮箱统一转成小写并去空格，真实业务应使用更严格的邮箱校验。
• 如果源文件来自上传流，需要先把流包装成文本迭代器，或改用 io.StringIO。

验证方式

运行后会输出一条合格记录，拒绝文件包含第 3 行年龄非整数和第 4 行年龄小于 18 两条原因。可以删除表头中的 email 列，确认程序在开始迭代前就报缺少列。