基于Node.js fast-csv包的CSV重复数据检测实现方案咨询

要实现phone字段的重复检测，核心难点在于重复校验是跨多行的逻辑——逐行校验阶段（validate回调）无法预知其他行的phone值。所以我们需要先收集所有解析完成的记录，再统一统计并标记重复项。

validateRows: (filePath, payload, validators) => new Promise((resolve, reject) => {
  const allRecords = []; // 新增：收集所有行的完整信息（数据、行号、原有错误）
  const fs = require('fs');
  const csv = require('fast-csv');

  fs.createReadStream(filePath)
    .pipe(csv.parse({
      headers: (headers) => mapHeaderToRelated(headers, payload),
      delimiter: ";",
      discardUnmappedColumns: true
    }))
    .validate((data, cb) => {
      // 保留原有单行校验逻辑，收集单行错误
      const errors = validators.reduce((err, func) => [...err, ...func(data)], []);
      if (errors.length > 0) {
        return cb(null, false, errors);
      }
      return cb(null, true);
    })
    .on("error", (error) => {
      console.log("There is some error");
      reject(error);
    })
    .on("data", (row, rowNumber) => {
      // 有效行存入allRecords，记录行号与空错误数组
      allRecords.push({
        data: row,
        rowNumber: rowNumber,
        reason: []
      });
    })
    .on("data-invalid", (row, rowNumber, reason) => {
      // 无效行存入allRecords，保留原有错误与行号
      allRecords.push({
        data: row,
        rowNumber: rowNumber,
        reason: reason
      });
    })
    .on("end", (rowCount) => {
      // 1. 统计每个phone的出现次数
      const phoneCounts = new Map();
      allRecords.forEach(record => {
        const phone = record.data.phone;
        if (phone) { // 忽略phone为空的情况，可根据需求调整逻辑
          phoneCounts.set(phone, (phoneCounts.get(phone) || 0) + 1);
        }
      });

      // 2. 重新筛选并生成有效/无效记录
      const validRecords = [];
      const invalidRecords = [];
      allRecords.forEach(record => {
        const phone = record.data.phone;
        const isDuplicate = phone && phoneCounts.get(phone) > 1;
        
        // 合并原有错误与重复错误
        let reasons = [...record.reason];
        if (isDuplicate) {
          reasons.push("Duplicate data");
        }

        if (reasons.length > 0) {
          invalidRecords.push({
            data: record.data,
            rowNumber: record.rowNumber,
            reason: reasons
          });
        } else {
          validRecords.push(record.data);
        }
      });

      console.log(`Parsed ${rowCount} rows. Valid Count: ${validRecords.length} Invalid Count: ${invalidRecords.length}`);
      resolve({ invalidRecords, validRecords });
    });
})

1. 新增allRecords数组：不管行是否通过原有校验，都统一存储行数据、行号、原有错误原因，为后续跨行的重复检测提供完整数据源。
2. 行号获取修正：利用fast-csvdata事件的第二个参数rowNumber（从1开始计数），补全有效行的行号记录，确保错误定位准确。
3. phone频次统计：使用Map高效统计每个phone值的出现次数，时间复杂度为O(n)，性能优于普通对象。
4. 最终记录分类：遍历所有记录，判断当前行phone是否重复，合并原有错误与重复错误后，将记录分别归入有效/无效数组。

针对你提供的示例CSV，运行后会生成符合预期的输出：

• 唯一的999999 phone记录被放入validRecords
• 两个123456的记录被标记为无效，错误原因包含["Duplicate data"]

内容的提问来源于stack exchange，提问作者Atakan Bal