合并240个CSV文件时遭遇_csv.Error: line contains NUL错误求助

错误原因 #

_csv.Error: line contains NUL 是因为目标CSV文件中混入了二进制空字符（\x00），常见诱因包括：

• 文件实际是二进制格式（比如误将.xlsx/.zip改名为.csv）
• 文件编码异常，读取时引入空字符
• 文件损坏或写入不完整

修复步骤及代码优化 #

核心改进方向 #

1. 过滤NUL字符：读取文件时主动剔除空字符，避免触发CSV解析错误
2. 逐文件编码检测：每个CSV文件单独检测编码，解决跨文件编码不兼容问题
3. 内存优化：边读边写合并文件，避免大文件占用过多内存
4. 修正表头逻辑：按表头首次出现顺序整理总表头，避免排序混乱

修改后的完整代码 #

import os
import csv
import chardet

directory_path = r"A:\FilesMerge"
output_file = os.path.join(directory_path, "merged_headers_and_data.csv")

# 收集所有唯一表头，按首次出现顺序存储
all_headers = []
header_set = set()

# 第一步：遍历所有CSV，收集所有表头
for filename in os.listdir(directory_path):
    if not filename.endswith(".csv"):
        continue
    file_path = os.path.join(directory_path, filename)
    
    # 检测当前文件编码
    with open(file_path, 'rb') as f:
        raw_data = f.read()
        encoding = chardet.detect(raw_data)['encoding'] or 'utf-8'
    
    # 读取文件并提取表头
    with open(file_path, 'r', encoding=encoding, errors="replace") as csvfile:
        # 过滤每行中的NUL字符
        filtered_content = (line.replace('\x00', '') for line in csvfile)
        reader = csv.reader(filtered_content)
        try:
            headers = next(reader)
            for header in headers:
                if header not in header_set:
                    header_set.add(header)
                    all_headers.append(header)
        except csv.Error as e:
            print(f"读取文件 {filename} 表头失败: {e}")
            continue

# 第二步：边读边写合并文件
with open(output_file, 'w', newline='', encoding='utf-8') as out_csv:
    writer = csv.writer(out_csv)
    writer.writerow(all_headers)
    
    # 构建总表头的索引映射，用于匹配各文件的列
    header_index_map = {header: idx for idx, header in enumerate(all_headers)}
    
    for filename in os.listdir(directory_path):
        if not filename.endswith(".csv"):
            continue
        file_path = os.path.join(directory_path, filename)
        
        # 再次检测编码（可缓存第一步结果提升效率）
        with open(file_path, 'rb') as f:
            raw_data = f.read()
            encoding = chardet.detect(raw_data)['encoding'] or 'utf-8'
        
        with open(file_path, 'r', encoding=encoding, errors="replace") as csvfile:
            filtered_content = (line.replace('\x00', '') for line in csvfile)
            reader = csv.reader(filtered_content)
            try:
                headers = next(reader)
                # 映射当前文件表头到总表头的索引
                current_header_indices = [header_index_map.get(h, -1) for h in headers]
                
                # 写入分隔行和文件名标识
                writer.writerows([[], [], [], [filename]])
                
                # 处理每行数据，映射到总表头对应的列
                for row in reader:
                    new_row = [''] * len(all_headers)
                    for idx, val in enumerate(row):
                        target_idx = current_header_indices[idx]
                        if target_idx != -1:
                            new_row[target_idx] = val
                    writer.writerow(new_row)
            except csv.Error as e:
                print(f"读取文件 {filename} 数据失败: {e}")
                continue

代码改进说明 #

• 分阶段处理：先收集所有表头，再逐文件写入数据，避免一次性加载所有数据到内存
• 错误隔离：单个文件读取失败时仅打印错误，不中断整个合并流程
• 编码兼容：每个文件单独检测编码，解决不同文件编码不一致的问题
• 数据映射：通过表头索引映射，确保不同结构的CSV数据能正确对应到合并文件的列

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