如何使用Python解析管道符分隔TXT文件并转换为CSV格式

核心逻辑 #

你的文件每行是值|列名交替排列的格式，且可能存在空值，直接用Python内置的csv模块处理即可，不需要额外依赖，自带的转义逻辑可以避免自己拼接字符串导致的csv格式错误。

完整代码（通用版，支持不同行存在不同列） #

import csv

# 配置参数
input_txt_path = "input.txt" # 替换成你的输入txt路径
output_csv_path = "output.csv" # 替换成你的输出csv路径
file_encoding = "utf-8" # 如果是gbk编码就改成gbk

# 第一步：先遍历所有行，收集所有出现过的列名
all_columns = set()
with open(input_txt_path, 'r', encoding=file_encoding) as f:
    for line in f:
        line = line.strip('\n')
        if not line:
            continue
        parts = line.split('|')
        # 取所有列名，也就是索引为奇数的元素
        for i in range(1, len(parts), 2):
            col_name = parts[i].strip()
            if col_name:
                all_columns.add(col_name)
# 转成列表方便后续按固定顺序写入
column_list = list(all_columns)

# 第二步：逐行处理写入csv
with open(input_txt_path, 'r', encoding=file_encoding) as in_f, \
     open(output_csv_path, 'w', newline='', encoding=file_encoding) as out_f:
    writer = csv.DictWriter(out_f, fieldnames=column_list)
    # 写表头
    writer.writeheader()
    for line in in_f:
        line = line.strip('\n')
        if not line:
            continue
        parts = line.split('|')
        row_dict = {}
        # 按值、列名交替提取
        for i in range(0, len(parts), 2):
            if i+1 >= len(parts):
                # 行元素数量为奇数，缺少对应列名，直接跳过
                break
            col_name = parts[i+1].strip()
            if not col_name:
                continue
            # 取值，保留原始内容，空值也正常写入
            value = parts[i].strip()
            row_dict[col_name] = value
        writer.writerow(row_dict)

简化版（所有行列顺序完全一致的场景） #

如果你能确定所有行的列顺序、列名都完全相同，不需要遍历两次文件，可以直接取第一行的列名做表头，处理速度更快：

import csv

input_txt_path = "input.txt"
output_csv_path = "output.csv"
file_encoding = "utf-8"

with open(input_txt_path, 'r', encoding=file_encoding) as in_f, \
     open(output_csv_path, 'w', newline='', encoding=file_encoding) as out_f:
    lines = [l.strip('\n') for l in in_f if l.strip('\n')]
    if not lines:
        exit()
    # 取第一行提取列名
    first_parts = lines[0].split('|')
    column_list = [first_parts[i].strip() for i in range(1, len(first_parts), 2) if first_parts[i].strip()]
    writer = csv.DictWriter(out_f, fieldnames=column_list)
    writer.writeheader()
    for line in lines:
        parts = line.split('|')
        row_dict = {}
        for idx, col in enumerate(column_list):
            value_idx = idx * 2
            if value_idx >= len(parts):
                value = ''
            else:
                value = parts[value_idx].strip()
            row_dict[col] = value
        writer.writerow(row_dict)

注意事项 #

• 代码默认保留字段的前后空格，如果需要去掉可以自行调整strip()的逻辑
• 空值会自动对应csv的空字段，不需要额外处理
• 如果文件超大（GB级别），建议用简化版避免两次IO读取占用过多资源

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