求助：pandas on_bad_line无法正确记录坏行所属文件名

我希望收集出现坏行的文件名及对应坏行内容，编写了函数将坏行写入bad_line1_日期.txt文件，但实际生成的文件里只打印了GCS存储桶的所有文件名，看不到预期的坏行错误信息。

现有代码 #

坏行收集函数 #

def badlines_collect(self, bad_line: list[str]) -> None:
    badline_lst.append(bad_line)
    today = date.today()
    todaytime = datetime.datetime.now().strftime("%Y%m%d")
    with open("bad_line1_{}.txt".format(todaytime), 'w') as fp:
        for line in badline_lst:
            fp.write("Today's date: " + str(today) + currentfile + ": {}\n".format(line))
    fp.close()
    print(badline_lst)
    return None

调用函数的代码 #

def getCSV(self, cur_publisher):
    """
    :return:
    """
    print(bucket_name + '/' + cur_publisher)
    dfm = pd.DataFrame()
    filename = list(self.bucket.list_blobs(prefix=cur_publisher))
    print(filename)
    for file_name in filename:
        if '.csv' in str(file_name.name):
            print("Crawling on File {} ......\n".format(file_name.name))
            currentfile = file_name.name
            print(currentfile)
            blop = self.bucket.blob(blob_name = "{}".format(file_name.name))
            data = blop.download_as_string()
            df = pd.read_csv(io.BytesIO(data), encoding='utf-8', sep=",", engine='python',
                             on_bad_lines=self.badlines_collect)
            if (df.count().sum()) > 0:
                df.insert(0, "filename", file_name.name)
                dfm = pd.concat([dfm, df], ignore_index=True)
                dfm = pd.concat([dfm, df], ignore_index=True)
                dfm = dfm.rename_axis(index='', columns="index")
                print(dfm)
            else:
                pass
                print("{} is empty \n".format(file_name.name))
        else:
            pass
    return self.stack

问题根源 #

1. 上下文传递错误：currentfile是getCSV循环中的局部变量，badlines_collect直接依赖该外部变量，会导致后续循环覆盖变量值，无法保证回调执行时变量对应正确文件。
2. 文件模式错误：使用'w'模式打开文件，每次调用badlines_collect都会清空之前的内容，最终只保留最后一次写入的信息。
3. 坏行内容格式化错误：直接输出bad_line列表对象，未转换为可读字符串；且写入逻辑重复遍历badline_lst，导致内容重复。
4. 全局变量风险：badline_lst作为全局变量，多文件处理时容易出现数据混乱。

修复方案 #

1. 修改坏行收集函数 #

将文件名作为参数传入，改用追加模式写入，同时格式化坏行内容：

from datetime import date, datetime

def badlines_collect(self, currentfile: str, bad_line: list[str]) -> None:
    # 使用实例变量存储坏行，避免全局变量冲突
    if not hasattr(self, 'badline_lst'):
        self.badline_lst = []
    self.badline_lst.append((currentfile, bad_line))
    
    today = date.today()
    todaytime = datetime.now().strftime("%Y%m%d")
    # 用追加模式('a')打开文件，保留历史内容
    with open(f"bad_line1_{todaytime}.txt", 'a', encoding='utf-8') as fp:
        # 将坏行列表转为逗号分隔的字符串，提升可读性
        fp.write(f"日期: {today} | 文件: {currentfile} | 坏行内容: {','.join(bad_line)}\n")
    print(f"已记录坏行：文件[{currentfile}]，内容[{bad_line}]")

2. 绑定回调参数 #

使用functools.partial将当前文件名绑定到回调函数，确保上下文正确：

from functools import partial
import io
import pandas as pd

def getCSV(self, cur_publisher):
    print(f"{bucket_name}/{cur_publisher}")
    dfm = pd.DataFrame()
    filename = list(self.bucket.list_blobs(prefix=cur_publisher))
    
    for file_name in filename:
        if '.csv' in file_name.name:
            print(f"Crawling on File {file_name.name} ......\n")
            currentfile = file_name.name
            blop = self.bucket.blob(blob_name=file_name.name)
            data = blop.download_as_string()
            # 绑定currentfile到回调函数
            df = pd.read_csv(io.BytesIO(data), encoding='utf-8', sep=",", engine='python',
                             on_bad_lines=partial(self.badlines_collect, currentfile))
            
            if df.count().sum() > 0:
                df.insert(0, "filename", file_name.name)
                dfm = pd.concat([dfm, df], ignore_index=True)
                dfm = dfm.rename_axis(index='', columns="index")
                print(dfm)
            else:
                print(f"{file_name.name} is empty \n")
    return self.stack

3. 可选优化：批量写入 #

如果坏行数量较多，可改为收集所有坏行后一次性写入，减少文件IO操作：

# 在getCSV函数末尾添加
if hasattr(self, 'badline_lst') and self.badline_lst:
    today = date.today()
    todaytime = datetime.now().strftime("%Y%m%d")
    with open(f"bad_line1_{todaytime}.txt", 'w', encoding='utf-8') as fp:
        for currentfile, bad_line in self.badline_lst:
            fp.write(f"日期: {today} | 文件: {currentfile} | 坏行内容: {','.join(bad_line)}\n")

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