如何定位CSV文件中非UTF-8格式行以完成AWS导入PostgreSQL操作

问题根因 #

你现有代码是一次性读取全量文件内容判断编码，触发解码错误时无法关联到具体行，改为逐行读取字节流校验即可定位问题。

可用解决方案 #

1. 定位错误行代码 #

直接运行以下脚本即可输出所有不符合UTF-8编码的行号和原始字节内容：

def find_bad_utf8_lines(file_path):
    # 二进制模式读取，避免提前解码报错
    with open(file_path, 'rb') as f:
        # 行号从1开始计数，符合日常使用习惯
        for line_num, line_bytes in enumerate(f, start=1):
            try:
                line_bytes.decode('utf-8')
            except UnicodeDecodeError:
                print(f"第 {line_num} 行存在非UTF-8字符，原始字节：{line_bytes!r}")

if __name__ == '__main__':
    # 替换为你的CSV文件路径
    find_bad_utf8_lines('filename.csv')

2. 自动修复编码生成新文件 #

如果不需要从源头修改，可直接运行以下脚本将文件转成标准UTF-8编码，兼容你预设的编码列表：

encodings = ['utf-8', 'windows-1250', 'windows-1252', 'latin-1']

def auto_fix_encoding(file_path, output_path):
    fixed_lines = []
    with open(file_path, 'rb') as f:
        for line_num, line_bytes in enumerate(f, start=1):
            decoded = None
            # 按优先级尝试解码
            for e in encodings:
                try:
                    decoded = line_bytes.decode(e)
                    break
                except UnicodeDecodeError:
                    continue
            # 所有编码都无法解码时，用�替换无效字符避免导入中断
            if decoded is None:
                decoded = line_bytes.decode('utf-8', errors='replace')
                print(f"第 {line_num} 行未匹配预设编码，已替换无效字符")
            fixed_lines.append(decoded)
    # 输出标准UTF-8编码的新文件
    with open(output_path, 'w', encoding='utf-8') as f:
        f.writelines(fixed_lines)

if __name__ == '__main__':
    auto_fix_encoding('filename.csv', 'fixed_utf8_filename.csv')

3. 额外适配建议 #

• 你提供的示例文本中存在"这类HTML实体字符，可导入前用html.unescape()方法转成正常双引号，避免入库后数据异常
• 后续使用Postgres COPY命令或者AWS DMS导入时，可直接使用转好的UTF-8文件，无需额外配置编码参数

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