Python如何在单个函数内使用两个enumerate语句定位CSV数据起止行

报错核心原因 #

• 传入函数的是已打开的文件对象，文件对象为流式读取，第一次遍历完所有行后指针停在文件末尾，第二次遍历无法读取到任何内容，导致结束行赋值逻辑永远不会触发，进而出现属性未定义/变量未正常赋值的报错
• 第一种写法的额外问题：给函数动态加属性的写法不规范，若结束行未匹配到，fileinfo.endFile 不会被创建，访问时直接触发属性不存在报错
• 第二种写法的额外问题：global 声明建议统一放在函数开头，虽然不属于语法强制，但部分Python解释器可能因声明位置问题识别异常，叠加结束行未赋值成功的问题，就会触发名称错误

正确实现方案 #

方案1：一次遍历同时定位起止行（最优，效率最高） #

无需两次读取文件，单次遍历即可同时拿到起止行号，完全规避文件指针问题，且支持提前退出遍历减少IO消耗

def get_data_boundary(file_path, start_flag='eventID', end_flag='The Line Listing is wrong'):
    begin_line = 0
    end_line = 0
    # 函数内部管理文件打开关闭，避免外部传入文件对象的状态异常问题
    with open(file_path, 'r', encoding='utf-8') as f:
        for line_num, line_content in enumerate(f, 1):
            if start_flag in line_content and begin_line == 0:
                begin_line = line_num
            if end_flag in line_content and end_line == 0:
                end_line = line_num
            # 两个标识都找到后提前退出，无需读完全部文件
            if begin_line != 0 and end_line != 0:
                break
    return begin_line, end_line

# 调用示例
beginFile, endFile = get_data_boundary("testbooklet.csv")
print(beginFile, endFile)

方案2：拆分两次查找（适配特殊业务场景） #

如果业务要求必须拆分起止行查找逻辑，每次遍历前需将文件指针移回开头：

def get_data_boundary(file_path, start_flag='eventID', end_flag='The Line Listing is wrong'):
    begin_line = 0
    end_line = 0
    with open(file_path, 'r', encoding='utf-8') as f:
        # 查找起始行
        for line_num, line_content in enumerate(f, 1):
            if start_flag in line_content:
                begin_line = line_num
                break
        # 指针重置到文件开头
        f.seek(0)
        # 查找结束行
        for line_num, line_content in enumerate(f, 1):
            if end_flag in line_content:
                end_line = line_num
                break
    return begin_line, end_line

批量处理扩展 #

针对批量CSV导入合并的需求，可直接基于上述函数封装完整的读取逻辑：

import pandas as pd

def read_valid_csv(file_path):
    begin, end = get_data_boundary(file_path)
    # 计算有效数据行数：结束行是错误提示行，不需要读取
    valid_rows = end - begin
    # 直接读取有效区间的内容，自动识别表头
    df = pd.read_csv(file_path, skiprows=begin-1, nrows=valid_rows-1)
    return df

# 批量处理示例
csv_file_list = ["test1.csv", "test2.csv", "test3.csv"]
all_valid_data = []
for csv_path in csv_file_list:
    df = read_valid_csv(csv_path)
    all_valid_data.append(df)

# 合并所有数据后可直接导出Excel或写入数据库
merged_df = pd.concat(all_valid_data, ignore_index=True)
merged_df.to_excel("merged_output.xlsx", index=False)

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