构建 CSV 导入校验管线

项目概述

本项目实现 CSV 导入校验管线：流式解析大文件，逐行按 schema 校验并累计带行号的类型化错误，暂存合格行，仅当错误数在阈值内时才以一次事务提交。默认全有或全无，因为半完成导入远比被拒绝的批次更难推理。

架构与关键决策

流式解析器一次产出一条记录，校验器累积错误而不在首个坏行中止，错误带行号、列与错误码，使报告能指向具体单元格。合格行写入按批次 id 归集的暂存表，提交前目标表不受影响，重导还可与上次尝试对账而非静默重复。阈值显式配置：零错误表示严格，正数容忍已知比例的坏行，把严格程度集中到一处。

实施与测试

导入混合有效与无效行的文件，断言接收数量与错误行号，尤其固定带表头时的行号偏移；错误数超阈值的批次断言目标表未变；以含逗号与换行的引号字段断言解析正确；重复导入同批次 id 断言为对账而非重复。

限制与后续

• 校验基于 schema 而非语义，不解析跨表引用。
• 管线单进程，超大文件受限于单机吞吐。
• 无增量续传，被拒批次须从头重导。

代码示例

pnpm init
pnpm add csv-parse pg
pnpm add -D typescript tsx vitest

csv-import-pipeline/
├── src/parse.ts
├── src/validate.ts
├── src/stage.ts
├── src/commit.ts
├── tests/import.test.ts
└── package.json

import { parse } from 'csv-parse';
input.pipe(parse({ columns: true, skip_empty_lines: true })).on('data', handleRow);

if (!isEmail(row.email)) errors.push({ line, field: 'email', value: row.email });

if (batch.length >= 500) { await insertBatch(batch); batch = []; }

if (errors.length > maxErrors) return rollback(batchId);

res.json({ batchId, accepted, rejected: errors.length, errors: errors.slice(0, 100) });

pnpm vitest run