使用Python修复列数异常的CSV文件，解决Teradata导入相关问题

一、先解决UnicodeDecodeError读取中断问题 #

脏数据常包含非标准编码字符，直接用utf-8读取会触发解码错误，可通过以下两种方式规避：

• 用latin-1编码读取（兼容所有字节，后续再清理无效字符）
• 给utf-8读取加错误替换参数

示例代码：

import csv

# 方式1：用latin-1读取
with open('dirty_data.csv', 'r', encoding='latin-1') as f:
    reader = csv.reader(f)
    rows = list(reader)

# 方式2：utf-8加错误替换（替换无法解码的字符）
# with open('dirty_data.csv', 'r', encoding='utf-8', errors='replace') as f:
#     reader = csv.reader(f)
#     rows = list(reader)

二、定位并修复列数不一致的行 #

遍历所有行，检查列数是否等于目标66列，记录异常行号并修复（修复逻辑可根据实际业务调整，以下为通用方案）：

target_col_count = 66
bad_row_records = []

for row_num, row in enumerate(rows, start=1):
    current_col_count = len(row)
    if current_col_count != target_col_count:
        bad_row_records.append(f"行号{row_num}：实际列数{current_col_count}（目标66）")
        # 修复逻辑
        if current_col_count < target_col_count:
            # 列数不足，补空值填充
            rows[row_num-1] = row + [''] * (target_col_count - current_col_count)
        else:
            # 列数过多，假设是字段内误加逗号，合并多余列到最后一个字段
            merged_row = row[:target_col_count-1] + [','.join(row[target_col_count-1:])]
            rows[row_num-1] = merged_row

# 打印所有列数异常行
print("列数异常行记录：")
for record in bad_row_records:
    print(record)

三、修复BTEQ解析失败的特殊字符 #

BTEQ对换行符、制表符、未转义的双引号敏感，需统一清洗：

def clean_bteq_chars(cell):
    cell_str = str(cell)
    # 替换换行、回车、制表符为空格
    cell_str = cell_str.replace('\n', ' ').replace('\r', ' ').replace('\t', ' ')
    # 转义双引号（CSV用双引号包裹字段时，字段内的双引号需转义为两个）
    if '"' in cell_str:
        cell_str = cell_str.replace('"', '""')
    return cell_str

# 清洗所有单元格
cleaned_rows = []
for row in rows:
    cleaned_row = [clean_bteq_chars(cell) for cell in row]
    cleaned_rows.append(cleaned_row)

四、保存清洗后的CSV并验证 #

保存时启用全字段引号包裹，避免BTEQ解析歧义，同时再次验证列数：

# 保存清洗后的文件
with open('cleaned_data.csv', 'w', encoding='utf-8', newline='') as f:
    # QUOTE_ALL确保所有字段用双引号包裹，避免BTEQ解析错误
    writer = csv.writer(f, quoting=csv.QUOTE_ALL)
    writer.writerows(cleaned_rows)

# 验证清洗结果
print("\n清洗后列数验证：")
with open('cleaned_data.csv', 'r', encoding='utf-8') as f:
    reader = csv.reader(f)
    for row_num, row in enumerate(reader, start=1):
        if len(row) != target_col_count:
            print(f"行号{row_num}仍存在列数异常")

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