Python自定义解析不规范文本文件为CSV教程-Python教程-PHP中文网

Python自定义解析不规范文本文件为CSV教程

本教程详细介绍了如何使用python处理格式不规范、以空格作为分隔符的文本文件，并将其转换为标准的csv格式。由于此类文件分隔符不一致且可能存在字段内部空格，标准库方法往往失效。文章通过自定义正则表达式解析函数，区分字段分隔符与内容空格，逐步实现文件读取、头部解析、数据行处理及最终csv输出，并强调了解决方案的数据依赖性及优化考量。

引言：处理不规范文本数据的挑战

在数据处理过程中，我们经常会遇到格式不规范的文本文件。这些文件可能使用不一致的空格数量作为字段分隔符，甚至在字段内容内部也包含空格，这使得传统的 pandas.read_csv 等方法难以准确解析。例如，当尝试使用 sep='\t' 或 sep=r"\s{2,}" 时，可能无法正确识别列边界，导致数据错位。对于这类“坏文件”，需要采用更精细的自定义解析策略。

核心思路：基于正则表达式的自定义解析

处理不规范文本文件的核心在于精确识别哪些空格序列是字段分隔符，哪些是字段内容的一部分。本教程将展示如何通过逐行读取文件，并利用正则表达式结合自定义替换函数（re.sub 与 replfunc）来解决这一问题。其基本思想是：

逐行读取：将文件视为一系列文本行。
区分行类型：识别文件头、空白行和数据行。
精确替换分隔符：对于数据行，通过分析不同长度的空格序列，将其替换为统一的制表符（\t）作为分隔符。
按制表符分割：使用替换后的制表符作为分隔符来分割字段。
构建结构化数据：将解析出的字段存储为列表的列表，最终转换为CSV。

实现步骤

1. 读取文件与初步处理

首先，我们需要打开并逐行读取文本文件。为了确保行尾字符不影响后续处理，通常会移除每行的换行符。

import sys
import re
import pandas as pd

def parse_bad_txt_to_csv(input_filepath, output_filepath):
    table = []
    with open(input_filepath, 'r', encoding='utf-8') as f:
        lines = f.readlines()

    for i, line in enumerate(lines):
        line = line.rstrip('\n') # 移除行尾换行符

        # ... 后续解析逻辑 ...

登录后复制

2. 解析文件头

通常，文件头（第一行）的字段分隔符可能相对一致，或者其结构可以通过简单的多空格分割来处理。

立即学习“Python免费学习笔记（深入）”；

        if i == 0:
            # 假设标题行中没有内部空格，可以直接按两个或更多空格分割
            header_row = re.split(r' {2,}', line)
            table.append(header_row)
            continue

登录后复制

3. 处理空白行

文件头和数据行之间可能存在空白行，这些行可以直接跳过。

        if not line.strip(): # 检查是否为空白行
            continue

登录后复制

4. 处理数据行：关键的空格分隔符识别

这是整个解析过程中最复杂也是最核心的部分。我们需要定义一个替换函数 replfunc，它将根据匹配到的空格序列的长度和上下文，决定是将其视为字段分隔符还是字段内容的一部分。

        def replfunc(mo):
            L = len(mo.group(0)) # 获取匹配到的空格序列长度

            # 特殊情况处理：例如，某些字段内部的两个空格应被保留为一个空格
            # 示例数据中 "Rejected at  level." 内部有两个空格，应视为一个内部空格
            (start, end) = mo.span()
            if L == 2:
                if (
                    line[:start].endswith('Rejected at')
                    and
                    line[end:].startswith('level.')
                ):
                    return ' ' # 替换为单个空格，作为字段内容

            # 其他情况：根据空格长度判断其作为字段分隔符的含义
            # 这些长度是根据具体数据模式观察得出的经验值
            if L < 2:
                # 理论上不应匹配到小于2个空格的序列，除非是特殊情况
                return mo.group(0) # 保留原样或抛出错误

            elif 2 <= L <= 12: # 较短的空格序列可能代表一个字段分隔符
                return '\t'
            elif L == 17: # 特定长度的空格序列可能代表多个空字段
                return '\t\t'
            elif L == 43:
                return '\t\t\t'
            elif L == 61:
                return '\t\t\t\t\t'
            elif L == 120:
                return '\t'
            elif L == 263:
                return '\t'
            else:
                # 如果遇到未预期的空格长度，可以返回原始空格或标记
                print(f"Warning: Unhandled space length {L} at line {i+1}")
                return '\t' # 默认替换为制表符

        # 使用 replfunc 替换所有两个或更多空格的序列
        tabbed_line = re.sub(r'\s{2,}', replfunc, line)
        row = tabbed_line.split('\t')
        table.append(row)

登录后复制

解析逻辑详解：

巧文书

巧文书是一款AI写标书、AI写方案的产品。通过自研的先进AI大模型，精准解析招标文件，智能生成投标内容。

查看详情

replfunc(mo)：这是一个回调函数，当 re.sub 找到一个匹配项时会被调用。mo 是匹配对象。
L = len(mo.group(0))：获取当前匹配到的空格序列的长度。
内部空格处理：if L == 2 的条件是一个关键的启发式判断。在示例数据中，"Rejected at level." 中的两个空格应被视为字段内容的一部分。通过检查前后文 (line[:start].endswith('Rejected at') 和 line[end:].startswith('level.'))，可以精确地将这对空格替换为单个空格，从而避免将其误判为字段分隔符。
分隔符识别：elif 语句块根据空格序列的长度将其映射为单个或多个制表符。例如，L == 17 被映射为 \t\t，这意味着原始文本中的17个空格实际上分隔了两个空字段。这些长度值是高度依赖于具体数据模式的，需要根据实际文件内容进行观察和调整。
默认处理：else 分支用于捕获所有未明确定义的空格长度。在实际应用中，这里可能需要更精细的日志记录或更智能的默认行为。

5. 构建结构化数据并转换为CSV文件

解析完所有行后，table 变量将包含一个列表的列表，每个内部列表代表一行数据。我们可以利用 pandas.DataFrame 将其转换为数据框，然后轻松导出为CSV。

    # 转换为 Pandas DataFrame
    # 确保所有行具有相同的列数，不足的用空字符串填充
    max_cols = max(len(row) for row in table)
    padded_table = [row + [''] * (max_cols - len(row)) for row in table]

    df = pd.DataFrame(padded_table[1:], columns=padded_table[0]) # 第一行为列名

    # 导出为CSV文件
    df.to_csv(output_filepath, index=False, encoding='utf-8')

    print(f"文件 '{input_filepath}' 已成功转换为 '{output_filepath}'。")

# 示例使用
# 假设你的不规范文本文件名为 'input.txt'
# parse_bad_txt_to_csv('input.txt', 'output.csv')

登录后复制

完整示例代码

import sys
import re
import pandas as pd

def parse_bad_txt_to_csv(input_filepath, output_filepath):
    """
    解析不规范的空格分隔文本文件，并将其转换为CSV文件。

    参数:
        input_filepath (str): 输入文本文件的路径。
        output_filepath (str): 输出CSV文件的路径。
    """
    table = []

    try:
        with open(input_filepath, 'r', encoding='utf-8') as f:
            lines = f.readlines()
    except FileNotFoundError:
        print(f"错误: 文件 '{input_filepath}' 未找到。")
        return
    except Exception as e:
        print(f"读取文件时发生错误: {e}")
        return

    for i, line in enumerate(lines):
        line = line.rstrip('\n') # 移除行尾换行符

        if i == 0:
            # 第一行作为文件头，假设其字段分隔符相对一致
            header_row = re.split(r' {2,}', line)
            table.append(header_row)
            continue

        if not line.strip(): # 检查是否为空白行
            continue

        # 定义替换函数，用于将不同长度的空格序列转换为制表符或单个空格
        def replfunc(mo):
            L = len(mo.group(0)) # 获取匹配到的空格序列长度

            # 特殊情况处理：例如，"Rejected at  level." 中的两个空格应被保留为一个空格
            (start, end) = mo.span()
            if L == 2:
                if (
                    line[:start].endswith('Rejected at')
                    and
                    line[end:].startswith('level.')
                ):
                    return ' ' # 替换为单个空格，作为字段内容

            # 其他情况：根据空格长度判断其作为字段分隔符的含义
            # 这些长度是根据具体数据模式观察得出的经验值，需要根据实际文件调整
            if L < 2:
                # 理论上不应匹配到小于2个空格的序列，除非是特殊情况
                # 这里可以根据需要进行更严格的错误处理或保留原始匹配
                return mo.group(0) 
            elif 2 <= L <= 12:
                return '\t'
            elif L == 17:
                return '\t\t'
            elif L == 43:
                return '\t\t\t'
            elif L == 61:
                return '\t\t\t\t\t'
            elif L == 120:
                return '\t'
            elif L == 263:
                return '\t'
            else:
                print(f"警告: 第 {i+1} 行发现未处理的空格长度 {L}。默认替换为制表符。")
                return '\t' # 默认替换为制表符，可能需要根据实际情况调整

        try:
            tabbed_line = re.sub(r'\s{2,}', replfunc, line)
            row = tabbed_line.split('\t')
            table.append(row)
        except Exception as e:
            print(f"处理第 {i+1} 行时发生错误: {e}。原始行: '{line}'")
            continue

    if not table:
        print("警告: 未解析到任何数据。")
        return

    # 转换为 Pandas DataFrame
    # 确保所有行具有相同的列数，不足的用空字符串填充
    max_cols = max(len(row) for row in table)
    padded_table = [row + [''] * (max_cols - len(row)) for row in table]

    # 第一行为列名，其余为数据
    if len(padded_table) > 1:
        df = pd.DataFrame(padded_table[1:], columns=padded_table[0])
    else: # 只有头部或没有数据行
        df = pd.DataFrame(columns=padded_table[0])

    # 导出为CSV文件
    try:
        df.to_csv(output_filepath, index=False, encoding='utf-8')
        print(f"文件 '{input_filepath}' 已成功转换为 '{output_filepath}'。")
    except Exception as e:
        print(f"导出CSV文件时发生错误: {e}")

# 假设你的不规范文本文件名为 'input.txt'，输出文件名为 'output.csv'
# 为了运行此代码，请创建一个名为 'input.txt' 的文件，并将问题中的数据粘贴进去。
# 然后取消注释下面一行并运行。
# parse_bad_txt_to_csv('input.txt', 'output.csv')

登录后复制

注意事项与优化

数据依赖性与通用性：
- 本教程中的 replfunc 逻辑是高度依赖于提供的示例数据模式的。尤其是不同长度空格序列映射到 \t 或 \t\t 的规则 (L == 17, L == 43 等) 是根据特定数据观察得出的。
- 在处理不同的不规范文本文件时，您需要仔细检查文件内容，分析空格分隔符的模式，并相应地调整 replfunc 中的条件。这通常需要手动检查几行数据，找出不同字段分隔符对应的空格长度。
- 对于更复杂或更不规则的文件，可能需要更高级的启发式算法，例如基于列对齐的算法，或者结合机器学习来识别字段边界。
鲁棒性与错误处理：
- 在实际应用中，应增加更全面的错误处理机制，例如 try-except 块来捕获文件读取、行解析或数据转换过程中可能出现的异常。
- 当 replfunc 遇到未预期的空格长度时，目前的实现会打印警告并默认替换为 \t。更健壮的做法可能是记录原始行和错误信息，或者根据业务需求采取其他策略。
性能考量：
- 对于非常大的文件，逐行读取和正则表达式操作可能会消耗较多内存和CPU。可以考虑使用生成器 (yield) 来逐块处理文件，避免一次性将整个文件读入内存。
- 如果文件非常大且模式复杂，可能需要考虑使用 C/C++ 扩展或更专业的文本解析库来优化性能。

总结

将不规范的文本文件转换为结构化的CSV格式是一项常见但可能具有挑战性的任务。当标准库函数无法满足需求时，自定义Python脚本结合正则表达式提供了一个强大而灵活的解决方案。通过精确识别并替换字段分隔符，我们可以有效地从混乱的文本数据中提取出有用的信息。然而，这种方法的有效性高度依赖于对具体数据模式的理解和细致的规则定义，因此在实际应用中需要仔细分析和调整。

以上就是Python自定义解析不规范文本文件为CSV教程的详细内容，更多请关注php中文网其它相关文章！