使用Pandas将循环中报错的行按格式追加写入CSV文件

开发CSV数据逐行入库函数时，需要将触发数据库专属错误（如完整性约束冲突）的对应行写入错误记录CSV文件。现有代码整体运行正常，但错误行写入逻辑存在格式异常：

• 预期格式：首行写入表头，后续每行横向存储单条错误记录的所有字段，以追加模式写入
• 实际输出：单条记录的每个字段单独占一行，且附带pandas默认行索引，每次写入会覆盖之前的错误记录

原有错误写入代码段：

error_file = pd.DataFrame(data)
error_file.to_csv('error.csv')

• 传入pd.DataFrame()的data是一维列表，pandas会默认将列表中每个元素解析为单独行，导致字段纵向排列
• 调用to_csv时未指定index=False，默认导出行索引
• 未指定文件写入模式，默认使用覆盖写入模式，之前记录的错误行会被新内容覆盖
• 没有单独处理表头写入逻辑，无法输出标准CSV的表头行

核心调整逻辑 #

• 进入逐行处理循环前，先判断错误文件是否存在，不存在则先写入表头，避免后续重复写入表头
• 构造错误记录DataFrame时，将单行数据包装为二维列表，同时指定列名匹配原CSV表头
• 写入错误文件时关闭索引导出，使用追加模式，追加写入时不再重复写入表头
• 也可以直接用Python标准库csv实现错误写入，减少pandas依赖，性能更稳定

修正后的完整函数代码 #

import os
import csv
import pandas as pd
# 注：db_connection、logging、DatabaseError需提前导入/定义

def insert_query(file):
    conn = db_connection()[0]
    df = pd.read_csv(file)
    error_file_path = 'error.csv'
    try:
        with open(file, 'r') as f:
            logging.critical('Attempting to insert data into table:')
            reader = csv.reader(f)
            columns = next(reader)
            query = '[insert query]'
            success = 0
            cursor = conn.cursor()
            num_row = df.shape[0]

            # 初始化错误文件，首次创建时写入表头
            if not os.path.exists(error_file_path):
                pd.DataFrame(columns=columns).to_csv(error_file_path, index=False)

            for data in reader:
                if num_row > 1:
                    try:
                        cursor.execute(query, data)
                        conn.commit()
                        logging.warning('Record inserted successfully')
                        success += 1
                        logging.critical(str(success) + ' rows inserted successfully')
                    except DatabaseError as e:
                        err, = e.args
                        logging.critical('Error: ' + str(data) + ' Error-Message:' + str(err.message))
                        # 构造单行数据的DataFrame，匹配原表列名
                        error_row = pd.DataFrame([data], columns=columns)
                        # 追加模式写入，关闭索引，不重复写入表头
                        error_row.to_csv(error_file_path, mode='a', index=False, header=False)
            logging.critical(str(success) + ' out of ' + str(num_row) + ' rows were inserted successfully')
            cursor.close()
            conn.close()
    except Exception as e:
        logging.critical('Encountered exception ' + str(e))

可选：无pandas依赖的轻量写入实现 #

如果不想在错误写入环节依赖pandas，可以直接用内置csv模块处理，避免不必要的 overhead：

# 初始化错误文件阶段替换为
if not os.path.exists(error_file_path):
    with open(error_file_path, 'w', newline='', encoding='utf-8') as err_f:
        err_writer = csv.writer(err_f)
        err_writer.writerow(columns)

# 异常捕获内的写入逻辑替换为
with open(error_file_path, 'a', newline='', encoding='utf-8') as err_f:
    err_writer = csv.writer(err_f)
    err_writer.writerow(data)

修复后写入的错误CSV格式符合预期，示例输出：

FName,LName,Address,Phone
john,smith,10 help st,4441342
tom,jackson,22 test rd,4445678

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