Python批量清洗CSV文件首行异常内容的实现方法

待处理文件分类规则 #

所有待处理CSV共分三类，表头字段无统一规则，清洗要求为删除无效首行、完整保留有效表头与全部数据，正常文件不做修改：

• 类型1：首行为单独逗号字符，第二行为有效表头
• 类型2：首行为sep=,分隔符标识行，第二行为有效表头
• 类型3：首行即为正常表头，无异常内容

核心实现逻辑 #

不直接用pandas默认参数全量读取文件，先预读首行做特征匹配，再确定读取参数，避免错读数据结构：

1. 遍历目标路径下所有CSV文件，跳过xls等其他格式文件
2. 以纯文本模式打开单个CSV，仅读取第一行原始内容，去除首尾换行、空白字符后做特征匹配
3. 若首行匹配两类异常特征，读取文件时指定header=1（从第二行开始识别表头）；若为正常首行，使用默认header=0参数读取
4. 读取完成后按原结构写回文件，不额外增加索引列、不修改字段顺序

完整实现代码 #

基于现有pathlib、pandas、time、argparse框架修改，无额外依赖：

from pathlib import Path
import pandas as pd
import time
import argparse

parser = argparse.ArgumentParser(description='CSV首行自动清洗工具')
parser.add_argument('path', help='指定待处理的文件或文件夹路径')

def clean_single_csv(fn: Path):
    # 预读首行判断表头位置
    header_row = 0
    with open(fn, 'r', encoding='utf-8', errors='ignore') as f:
        first_line = f.readline().strip()
        # 匹配两类异常首行特征
        if first_line == ',' or first_line == 'sep=,':
            header_row = 1
    # 按正确的表头位置读取文件，保留原始字段类型不做自动转换
    df = pd.read_csv(fn, header=header_row, dtype='unicode', index_col=False)
    # 写回文件，不保留pandas自动生成的行索引
    df.to_csv(fn, index=False)
    print(f"已处理完成：{fn.name}")

def main(path_name: Path):   
    if path_name.is_file():
        fnames = [path_name]
    else:
        fnames = list(Path(path_name).glob("*.csv")) + list(Path(path_name).glob("*.xls"))       
    for fn in fnames:
        if fn.suffix.lower() == '.csv':
            clean_single_csv(fn)
        # xls文件可按实际需求扩展处理逻辑
        else:
            print(f"跳过非CSV文件：{fn.name}")
  
if __name__== "__main__":
    start = time.time()
    args = parser.parse_args()
    path = Path(args.path)
    main(path)
    print(f"处理总耗时：{time.time()-start:.2f}秒")

使用说明 #

• 运行方式和原有框架一致，命令行传入目标文件/文件夹路径即可，例如python clean_csv.py ./target_folder
• 重要提醒：代码默认直接覆盖原文件，正式运行前建议先备份原始数据，或修改to_csv的保存路径到单独输出目录，避免数据丢失
• 若遇到编码报错，可将代码中encoding='utf-8'替换为encoding='gbk'或其他对应文件编码

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