Python如何统计CSV每行分隔符数并截断超13列冗余内容

问题说明 #

人工录入的CSV数据共包含13个业务列，整体录入错误率不足10%，但存在两类影响数据库写入的异常：

• 字段中夹带非法特殊字符
• 违规录入额外内容（如第二个邮箱）、甚至在字段中插入分隔符;，导致单行列数超过13列，加载时出现字段错位
  清洗规则要求自动截断第13列之后的所有内容，允许截断导致的部分数据丢失，同时过滤非法特殊字符。

正常CSV样例 #

TypeOfEntry;Schoolid;Schoolyear;Grade;Classname;ClassId;firstname;lastname;Gender;nationality;Street;housenumber;Email;

;;;;;;;;;;;;;

U;98645;2022;4;4AG;59845;John;Bizley;Male;United Kingdom;Canterburrystreet; 15a; Jb2004@hotmail.com;
U;98645;2022;4;4AG;59847;Alice;Schmidt;Female;United Kingdom;Milton street; 2/3; alice.schmidt@hotmail.com;

异常行样例 #

U;98645;2022;5;6CD;59845;Billy;Snow;Male;United Kingdom;Freedom street; 2a; BillyS@gmail.com;Billysnow2004@hotmail.com;

实现思路 #

• 不直接使用pandas加载原始异常CSV，避免列数不匹配导致的加载报错或字段错位
• 逐行读取原始文件文本，按分隔符;切分后直接保留前13列对应内容，丢弃所有超出范围的字段，从根源解决列数超限问题
• 预处理完成的文本再加载为DataFrame，执行原有特殊字符过滤逻辑
• 补充字段首尾空格清理，优化导出数据格式

完整可运行代码 #

import pandas as pd
from datetime import datetime
from io import StringIO

# 基础配置
TARGET_COL_COUNT = 13
INPUT_FILE_PATH = "schooldata.csv"
COL_NAMES = [
    'TypeOfEntry','Schoolid','Schoolyear','Grade','Classname','ClassId',
    'Firstname','Lastname','Gender','Nationality','Street','Housenumber','Email'
]

# 第一步：逐行预处理，截断超量列
processed_lines = []
with open(INPUT_FILE_PATH, 'r', encoding='utf-8') as f:
    for line in f:
        # 移除行尾换行符
        line_content = line.rstrip('\r\n')
        # 按分号切分字段
        field_parts = line_content.split(';')
        # 保留前13个业务列对应的分段（匹配CSV末尾带分号的格式特征），丢弃后续所有内容
        if len(field_parts) > TARGET_COL_COUNT + 1:
            field_parts = field_parts[:TARGET_COL_COUNT + 1]
        processed_lines.append(';'.join(field_parts))

# 将预处理后的文本转为内存文件对象供pandas读取
processed_csv_buffer = StringIO('\n'.join(processed_lines))

# 加载数据
data = pd.read_csv(processed_csv_buffer, sep=';', dtype=str)
# 强制只保留前13个业务列，兼容空行等异常格式
data = data.iloc[:, :TARGET_COL_COUNT]
data.columns = COL_NAMES

# 原有特殊字符过滤逻辑
data = data.convert_dtypes()
rep_chars = '°|^|!|"|\(|\)|\?'
rep_chars2 = r'\'|\`|\´|\*|#'
data = data.replace(rep_chars, '', regex=True)
data = data.replace(rep_chars2, '', regex=True)
data = data.replace('\+', '-', regex=True)

# 清理字段首尾多余空格
for col in data.columns:
    if pd.api.types.is_string_dtype(data[col]):
        data[col] = data[col].str.strip()

# 导出结果
export_date = datetime.now().strftime("%Y_%m_%d")
print(data.head())
print(data.dtypes)
data.to_csv(f'scoolexport_{export_date}.csv', sep=';', date_format='%Y%m%d', index=False)

效果说明 #

• 所有列数超过13的异常行都会被自动截断，比如样例中的双邮箱异常行，会自动丢弃第二个邮箱字段，保留前13列的正确结构
• 原有特殊字符过滤规则完全保留，不会破坏之前已生效的清洗逻辑
• 字段首尾多余空格会被自动清理，避免邮箱、地址类字段因为前后空格导致匹配失败

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