清理并高效读取含非结构化文本的CSV文件

# 创建一个示例文件
sample_data = """SAMPLE FILE LTD

STUDENT NUMBERS

INFO OF ALL STUDENTS No :  from 27-Mar-2023 00:00:00 to 04-Apr-2023 00:00:00 and  from 05-Oct-2023 00:00:00 to 13-Oct-2023 00:00:00

Student,id,add,div,rank
ABC,12,USA,A,1
DEF,13,IND,C,2
XYZ,14,UK,E,3
PQR,15,DE,F,4


 This is System generated report, and needs no signature.

 14-Oct-2023 18:14:12
"""

with open('students.csv', 'w') as f:
    f.write(sample_data)

print("示例文件 students.csv 已创建。")

Student,id,add,div,rank
ABC,12,USA,A,1
DEF,13,IND,C,2
XYZ,14,UK,E,3
PQR,15,DE,F,4

import pandas as pd

def get_rows_to_skip(file_name, header_keyword):
    """
    计算从文件开头到包含指定关键词的行之间的行数。
    """
    rows = 0
    with open(file_name, 'r') as file:
        while True:
            line = file.readline()
            if not line: # 文件结束
                return -1 # 表示未找到关键词
            if header_keyword in line:
                return rows
            rows += 1

def read_cleaned_csv_by_skiprows(file_name, header_keyword, expected_columns):
    """
    通过跳过指定行数来读取清理后的CSV文件。
    """
    skip_rows = get_rows_to_skip(file_name, header_keyword)
    if skip_rows == -1:
        print(f"错误：未在文件 '{file_name}' 中找到关键词 '{header_keyword}'。")
        return pd.DataFrame()

    # 读取CSV，跳过前导行
    df = pd.read_csv(file_name, skiprows=skip_rows)

    # 清理数据：移除全为空的行，这通常发生在文件末尾的冗余文本被读入后
    # 确保列名是预期的，并且数据行不包含NaN
    # 检查第一行是否是预期的列名，如果不是，则可能需要进一步处理
    if not all(col in df.columns for col in expected_columns):
        print(f"警告：读取的列名与预期不符。实际列名：{df.columns.tolist()}")
        # 尝试将第一行作为列名并重新读取，或者进行更复杂的清洗
        # 考虑到我们的header_keyword是'rank'，它在列名中，所以pd.read_csv会正确识别头部。
        pass

    # 移除所有列都为NaN的行，这有助于清理文件末尾的空行或无关文本
    df = df.dropna(how='all')

    # 进一步清理：如果某些列被读取为NaN，但它们应该是数据，这通常意味着文件末尾有额外文本
    # 我们可以根据关键列（如'rank'）来过滤掉无效数据行
    if 'rank' in df.columns:
        df = df[df['rank'].notna()]

    return df

# 预期列名，用于验证和清理
expected_columns = ['Student', 'id', 'add', 'div', 'rank']
df_skiprows = read_cleaned_csv_by_skiprows('students.csv', 'rank', expected_columns)
print("方法一：基于关键词跳过行读取结果:")
print(df_skiprows)
print("-" * 30)

import pandas as pd

def read_cleaned_csv_by_line_scan(file_name, header_start_string, expected_columns):
    """
    通过逐行扫描定位头部，然后读取清理后的CSV文件。
    """
    with open(file_name, 'r') as file:
        header_line = None
        # 逐行读取直到找到头部行
        for line in file:
            if line.strip().startswith(header_start_string):
                header_line = line.strip()
                break

        if header_line is None:
            print(f"错误：未在文件 '{file_name}' 中找到以 '{header_start_string}' 开头的头部行。")
            return pd.DataFrame()

        # 从找到的头部行解析列名
        column_names = header_line.split(',')

        # 将文件句柄的剩余部分传递给pd.read_csv
        # 使用 names 参数指定列名，因为我们已经读取了头部行
        df = pd.read_csv(file, names=column_names)

        # 清理数据：移除所有列都为NaN的行，这有助于清理文件末尾的空行或无关文本
        df = df.dropna(how='all')

        # 进一步清理：如果某些列被读取为NaN，但它们应该是数据，这通常意味着文件末尾有额外文本
        if 'rank' in df.columns:
            df = df[df['rank'].notna()]

        return df

# 使用 'Student' 作为头部行的起始字符串
df_line_scan = read_cleaned_csv_by_line_scan('students.csv', 'Student', expected_columns)
print("方法二：逐行扫描定位头部读取结果:")
print(df_line_scan)
print("-" * 30)

import pandas as pd

def read_cleaned_csv_by_full_read(file_name):
    """
    将整个文件读取为字符串，然后进行分割和清洗。
    """
    with open(file_name, 'r') as file:
        # 将整个文件读取为字符串，然后按换行符分割成行
        df_raw = pd.DataFrame(file.read().split('\n'))

    # 将单列DataFrame的每一行按逗号分割成多列
    # dropna() 用于移除文件开头和结尾的非CSV行，以及可能存在的空行
    df = df_raw[0].str.split(',', expand=True).dropna()

    # 将第一行（此时应该是实际的CSV头部）设置为列名
    # 并移除原始的第一行（因为现在它是列名）
    df.columns = df.iloc[0].values
    df = df[1:].reset_index(drop=True)

    # 再次清理，确保数据行中没有NaN
    df = df.dropna(how='any')

    return df

df_full_read = read_cleaned_csv_by_full_read('students.csv')
print("方法三：全文件读取与后处理读取结果:")
print(df_full_read)
print("-" * 30)