如何使用FastCSV识别不同输入schema并归一化多格式CSV数据

你的三个思路里，思路1的可扩展性最好，后续如果新增第三种winner格式只需要加对应的表头识别规则和行处理逻辑即可，不需要修改核心流处理代码。

核心实现逻辑 #

• 先定义统一的标准输出结构，不管输入是Winner还是Winner2格式，最终都输出该结构的实例，避免后续业务逻辑兼容多种类型。
• 利用@fast-csv/parse的表头钩子识别格式：parse方法的headers参数支持传入函数，在解析表头阶段就判断当前是哪种格式，同时直接把不同格式的表头映射成标准结构的字段名，从源头解决表头不一致的问题。
• 对应格式的行处理逻辑可配置，识别到格式后自动绑定对应的行转换方法，不需要在transform里写大量类型判断代码。

TypeScript + Node Streams 实现代码 #

import { parse, RowTransformCallback } from "@fast-csv/parse";
import { format } from "@fast-csv/format";

// 统一的标准输出结构
interface NormalizedWinner {
  id: string;
  name: string;
  prize: string;
  winTime: Date;
}

// 输入格式的配置映射
type CsvFormatConfig = {
  // 表头匹配规则，判断属于哪种格式
  headerMatch: (headers: string[]) => boolean;
  // 表头映射规则，把输入表头转成标准结构字段
  headerMap: Record<string, keyof NormalizedWinner>;
  // 行数据预处理逻辑，处理字段类型转换等
  rowProcessor: (row: Record<string, any>) => NormalizedWinner;
};

// 定义Winner格式配置
const winnerFormat: CsvFormatConfig = {
  headerMatch: headers => headers.includes("winner_id") && headers.includes("winner_name"),
  headerMap: {
    "winner_id": "id",
    "winner_name": "name",
    "award": "prize",
    "winning_time": "winTime"
  },
  rowProcessor: row => ({
    id: row.id,
    name: row.name,
    prize: row.prize,
    winTime: new Date(row.winTime)
  })
};

// 定义Winner2格式配置
const winner2Format: CsvFormatConfig = {
  headerMatch: headers => headers.includes("user_id") && headers.includes("user_name"),
  headerMap: {
    "user_id": "id",
    "user_name": "name",
    "prize_name": "prize",
    "create_time": "winTime"
  },
  rowProcessor: row => ({
    id: row.id,
    name: row.name,
    prize: row.prize,
    winTime: new Date(row.winTime)
  })
};

const ALL_FORMATS: CsvFormatConfig[] = [winnerFormat, winner2Format];

核心流处理代码：

// 动态匹配格式的处理逻辑
let currentFormat: CsvFormatConfig | null = null;

stream
  .pipe(parse({
    headers: (headers: string[]) => {
      // 匹配当前CSV对应的格式
      currentFormat = ALL_FORMATS.find(f => f.headerMatch(headers));
      if (!currentFormat) throw new Error("不支持的CSV格式");
      // 把输入表头映射成标准字段名
      return headers.map(h => currentFormat.headerMap[h] || h);
    }
  }))
  .transform((row, next: RowTransformCallback<NormalizedWinner, NormalizedWinner>) => {
    try {
      const processedRow = currentFormat.rowProcessor(row);
      next(null, processedRow);
    } catch (err) {
      next(err as Error);
    }
  })
  .pipe(format<NormalizedWinner, NormalizedWinner>({ headers: true }))
  .pipe(process.stdout)
  .on('error', reject)
  .on('end', (rowCount: number) => console.log(`Parsed ${rowCount} rows`));

RxJS实现方案 #

完全可以用RxJS实现，RxJS对异步流的控制更灵活，适合需要复杂逻辑编排的场景，核心逻辑和流处理一致：

1. 用from(NodeReadableStream)把CSV输入流转成Observable
2. 拆分表头行和数据行，先处理第一行判断格式
3. 后续数据行用对应格式的处理逻辑转成标准结构
4. 最终可以转成Node Stream输出，或者直接订阅处理结果

RxJS示例代码：

import { from, take, skip, map } from "rxjs";
import { parse } from "@fast-csv/parse";

const csv$ = from(stream.pipe(parse({ headers: false })));
let headerRow: string[];

// 先拿第一行表头判断格式
csv$.pipe(
  take(1),
  map(row => {
    headerRow = row as string[];
    const format = ALL_FORMATS.find(f => f.headerMatch(headerRow));
    if (!format) throw new Error("不支持的CSV格式");
    return format;
  })
).subscribe(format => {
  // 处理后续所有数据行
  csv$.pipe(
    skip(1), // 跳过表头行
    map(row => {
      // 先把行数据和表头映射成键值对
      const rowObj = Object.fromEntries(
        (row as string[]).map((val, idx) => [format.headerMap[headerRow[idx]] || headerRow[idx], val])
      );
      return format.rowProcessor(rowObj);
    })
  ).subscribe({
    next: normalizedRow => { /* 处理归一化后的数据 */ },
    error: reject,
    complete: () => console.log("解析完成")
  })
});

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