如何合并表头含差异、空值及重复的多份CSV文件？

需求说明 #

• 合并多份CSV文件，每份含数百列、数千行
• 表头多数重合但不全一致，存在空表头、未知重复表头
• 目标：仅保留所有文件的公共列，移除重复列，添加file（自定义文件昵称）和count（文件内行号）列

现有问题 #

使用原生csv模块合并时，出现三个问题：保留非公共列、重复写入表头、无法添加自定义标识列；而Pandas的read_csv会自动将重复表头重命名为x.1，不符合需求。

完整实现代码 #

import csv

def get_common_headers(filenames):
    headers_list = []
    for file in filenames:
        with open(file, 'r', newline='', encoding='utf-8') as f:
            reader = csv.reader(f)
            headers = next(reader)
            # 去重当前文件的表头，保留第一个出现的列
            unique_headers = []
            seen = set()
            for h in headers:
                if h not in seen:
                    seen.add(h)
                    unique_headers.append(h)
            headers_list.append(unique_headers)
    # 提取所有文件的公共表头，按第一个文件的表头顺序排列
    common_set = set(headers_list[0]).intersection(*headers_list[1:])
    common_headers = [h for h in headers_list[0] if h in common_set]
    return common_headers

def merge_csvs(filenames, output_file):
    common_headers = get_common_headers(filenames)
    final_headers = common_headers + ['file', 'count']
    
    with open(output_file, 'w', newline='', encoding='utf-8') as out_f:
        writer = csv.writer(out_f)
        # 仅写入一次最终表头
        writer.writerow(final_headers)
        
        for file in filenames:
            # 自定义文件昵称，这里取文件名，可根据需求修改
            file_nickname = file.split('/')[-1]
            with open(file, 'r', newline='', encoding='utf-8') as in_f:
                reader = csv.reader(in_f)
                headers = next(reader)
                # 建立表头到索引的映射，重复表头仅保留第一个出现的索引
                header_to_idx = {}
                for idx, h in enumerate(headers):
                    if h not in header_to_idx:
                        header_to_idx[h] = idx
                # 遍历数据行，行号从1开始计数
                row_count = 1
                for row in reader:
                    # 提取公共列内容，行长度不足时补空字符串
                    common_row = []
                    for h in common_headers:
                        if header_to_idx[h] < len(row):
                            common_row.append(row[header_to_idx[h]])
                        else:
                            common_row.append('')
                    # 添加自定义列
                    common_row.extend([file_nickname, row_count])
                    writer.writerow(common_row)
                    row_count += 1

# 调用示例
if __name__ == '__main__':
    # 替换为你的CSV文件列表
    target_files = ['data1.csv', 'data2.csv', 'data3.csv']
    merge_csvs(target_files, 'merged_result.csv')

关键处理说明 #

• 公共表头提取：先对单个文件的表头去重（保留首次出现的列），再计算所有文件的表头交集，同时按第一个文件的表头顺序排列，避免列顺序混乱
• 重复表头过滤：读取单个文件时，同一表头仅保留第一个出现的列索引，后续重复列直接忽略
• 非公共列移除：仅提取公共表头对应的列，自动过滤各文件独有的列
• 自定义列添加：file列默认使用文件名作为标识，可自行修改为自定义别名；count列记录每个文件内的行号（从1开始）
• 表头唯一写入：合并文件仅在开头写入一次最终表头，不会重复输出表头行

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