清理带有冗余文本的CSV文件：Pandas高效导入策略

import pandas as pd
import io

# 模拟一个包含冗余文本的CSV文件内容
sample_csv_content = """
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
"""

def read_clean_csv_by_header_detection(file_path_or_content, header_indicator="Student"):
    """
    通过检测特定头部指示器来清洗并读取CSV文件。

    Args:
        file_path_or_content (str): CSV文件的路径或包含CSV内容的字符串。
        header_indicator (str): 用于识别数据头部的关键字符串。

    Returns:
        pd.DataFrame: 清理后的数据DataFrame。
    """
    if isinstance(file_path_or_content, str) and '\n' in file_path_or_content:
        # 如果是字符串内容，使用io.StringIO模拟文件
        file_obj = io.StringIO(file_path_or_content)
    else:
        # 如果是文件路径，直接打开文件
        file_obj = open(file_path_or_content, 'r')

    try:
        # 逐行读取直到找到头部
        lines_to_skip = 0
        while True:
            line = file_obj.readline()
            if not line: # 文件结束仍未找到
                raise ValueError(f"未找到包含 '{header_indicator}' 的头部行。")
            if header_indicator in line:
                # 找到头部行，将该行作为列名
                column_names = [col.strip() for col in line.strip().split(',')]
                break
            lines_to_skip += 1

        # 使用pd.read_csv从当前文件位置开始读取
        # names参数用于指定列名，因为我们已经手动获取了它们
        df = pd.read_csv(file_obj, names=column_names)

        # 清理可能存在的尾部冗余行（例如，全NaN的行）
        df = df.dropna(how='all')

        return df
    finally:
        if file_obj is not None and not isinstance(file_obj, io.StringIO):
            file_obj.close()

# 示例调用
# 对于实际文件：
# df_cleaned = read_clean_csv_by_header_detection("students.csv", header_indicator="Student")

# 对于模拟内容：
df_cleaned = read_clean_csv_by_header_detection(sample_csv_content, header_indicator="Student")
print(df_cleaned)

import pandas as pd
import io

# 模拟文件内容
sample_csv_content = """
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
"""

def read_clean_csv_by_full_string_processing(file_path_or_content, header_pattern="Student,id"):
    """
    通过整体读取文件内容并使用Pandas字符串操作清洗CSV。

    Args:
        file_path_or_content (str): CSV文件的路径或包含CSV内容的字符串。
        header_pattern (str): 用于识别数据头部的完整字符串模式。

    Returns:
        pd.DataFrame: 清理后的数据DataFrame。
    """
    if isinstance(file_path_or_content, str) and '\n' in file_path_or_content:
        # 如果是字符串内容
        file_content = file_path_or_content
    else:
        # 如果是文件路径
        with open(file_path_or_content, 'r') as file:
            file_content = file.read()

    # 将整个文件内容按行分割，创建单列DataFrame
    df_raw = pd.DataFrame(file_content.split('\n'))

    # 清理空行
    df_raw = df_raw[df_raw[0].str.strip() != '']

    # 查找头部行的索引
    header_index = df_raw[df_raw[0].str.contains(header_pattern, na=False)].index[0]

    # 提取列名
    column_names = [col.strip() for col in df_raw.iloc[header_index, 0].split(',')]

    # 提取数据行，从头部下一行开始
    df_data = df_raw.iloc[header_index + 1:]

    # 将单列数据按逗号分割成多列
    df_processed = df_data[0].str.split(',', expand=True)

    # 设置列名
    df_processed.columns = column_names

    # 清理尾部可能存在的全NaN行
    df_processed = df_processed.dropna(how='all')

    return df_processed

# 示例调用
# 对于实际文件：
# df_cleaned_full = read_clean_csv_by_full_string_processing("students.csv", header_pattern="Student,id,add,div,rank")

# 对于模拟内容：
df_cleaned_full = read_clean_csv_by_full_string_processing(sample_csv_content, header_pattern="Student,id,add,div,rank")
print(df_cleaned_full)