Python实现多CSV文件合并：含表头校验与文件归档处理

这里有一个实用的Python脚本，完美匹配你需要的CSV批量处理流程——从扫描文件、表头校验、内容合并到完成文件归档，一步到位：

核心逻辑说明 #

• 自动扫描指定文件夹下所有.csv文件，按文件名排序保证处理顺序（比如file1.csv→file2.csv...）
• 第一个文件作为基准，提取表头并完整写入合并文件combined.csv
• 后续文件先校验表头与基准是否一致，一致则仅追加数据行（跳过自身表头），不一致则记录错误并跳过内容写入
• 处理完成的文件自动移动到父目录下的/done文件夹，避免重复处理

完整代码实现 #

import os
import shutil
import csv

def process_csv_folder(source_folder):
    # 1. 准备工作：创建done文件夹，获取所有CSV文件并排序
    done_folder = os.path.join(os.path.dirname(source_folder), "done")
    os.makedirs(done_folder, exist_ok=True)
    
    # 获取所有CSV文件并按文件名排序
    csv_files = [f for f in os.listdir(source_folder) if f.lower().endswith('.csv')]
    csv_files.sort()  # 确保按file1、file2...的顺序处理
    
    if not csv_files:
        print("未找到任何CSV文件")
        return
    
    combined_file_path = os.path.join(source_folder, "combined.csv")
    base_header = None
    
    # 2. 处理第一个CSV文件（作为基准）
    first_file = csv_files[0]
    first_file_path = os.path.join(source_folder, first_file)
    
    with open(first_file_path, 'r', encoding='utf-8') as infile, open(combined_file_path, 'w', encoding='utf-8', newline='') as outfile:
        reader = csv.reader(infile)
        writer = csv.writer(outfile)
        
        # 读取并保存基准表头
        base_header = next(reader)
        writer.writerow(base_header)
        
        # 写入第一个文件的剩余数据行
        writer.writerows(reader)
    
    # 移动第一个文件到done文件夹
    shutil.move(first_file_path, os.path.join(done_folder, first_file))
    print(f"已处理并归档：{first_file}")
    
    # 3. 处理剩余CSV文件
    for filename in csv_files[1:]:
        file_path = os.path.join(source_folder, filename)
        
        try:
            with open(file_path, 'r', encoding='utf-8') as infile:
                reader = csv.reader(infile)
                current_header = next(reader)
                
                # 校验表头是否与基准一致
                if current_header != base_header:
                    print(f"警告：{filename}表头与基准不一致，跳过内容写入")
                    shutil.move(file_path, os.path.join(done_folder, filename))
                    continue
                
                # 表头一致，追加数据行（跳过自身表头）
                with open(combined_file_path, 'a', encoding='utf-8', newline='') as outfile:
                    writer = csv.writer(outfile)
                    writer.writerows(reader)
            
            # 移动处理完成的文件
            shutil.move(file_path, os.path.join(done_folder, filename))
            print(f"已处理并归档：{filename}")
        
        except Exception as e:
            print(f"处理{filename}时出错：{str(e)}")
            # 出错时也移动文件，防止重复处理
            shutil.move(file_path, os.path.join(done_folder, filename))

# 使用示例：替换为你的CSV文件夹路径
if __name__ == "__main__":
    csv_folder = "./your_csv_folder"  # 这里改成你的CSV文件夹路径
    process_csv_folder(csv_folder)

关键细节说明 #

• 编码处理：脚本使用utf-8编码读取和写入文件，如果你的CSV是其他编码（比如gbk），可以修改encoding参数
• 文件排序：通过csv_files.sort()保证按文件名顺序处理，如果你需要按其他规则排序（比如修改时间），可以调整排序逻辑
• 错误处理：即使文件处理出错，也会将其移动到done文件夹，避免重复处理，同时打印错误信息方便排查
• 重复运行：如果脚本中途中断，再次运行时会自动跳过已经归档到/done的文件，只处理剩余的CSV文件

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