如何用Python程序化对齐CSV文件中各行的字段

手头有一组转换后的CSV格式老旧数据，存在各行字段格式不一致的问题：部分行长度相同则字段对应一致，但行长度不同时字段位置错乱，导致无法直接复用数据。数据示例如下：

data = """
30,1204,PO,71100,147130,I09,B10,OC,350,20105402
31,1221,PO,70400,147170,I09,B10,OC,500,20105402
32,1223,SI,70384,147122,I09,B10,OC,500,PN,3,BO,OI,20105402
33,1224,SI,70392,147032,I09,B10,OC,500,PN,1,BO,OI,20105402
34,1227,PO,70400,146430,I09,B10,PF,500,20105402
35,1241,PO,71100,146420,I09,B10,PF,500,20105402
36,1249,PO,71100,146000,I09,B10,SN,500,20105402
37,1305,PO,70400,146000,I09,B10,OC,500,20105402
38,1307,SI,70379,146041,I09,B10,OC,500,21,BH,1,BO,195,40,SW,20105402
39,1312,SD,70372,146062,I09,B10,OC,500,20105402
40,1332,SI,70334,146309,I09,B10,OC,500,PN,4,BO,OI,20105402
41,1332,SI,70334,146309,I09,B10,OC,500,PN,5,BO,OI,20105403
42,1333,SI,70333,146324,I09,B10,OC,500,PN,2,BO,OI,20105403
43,1334,SI,70328,146348,I09,B10,OC,500,PN,1,BO,OI,20105403
44,1335,SI,70326,146356,I09,B10,OC,500,PN,1,BO,OI,20105403
45,1336,SI,70310,146424,I09,B10,OC,500,PN,1,BO,OI,20105403
46,1338,SI,70302,146457,I10,B10,OC,500,PN,1,BO,OI,20105403
47,1338,SI,70301,146464,I10,B10,OC,500,PN,1,BO,OI,20105403
48,1340,SI,70295,146503,I10,B10,OC,500,PN,8,BO,OI,20105403
49,1405,LD,2,70119,148280,I10,B10,OC,0000,20105403
01,1024,LA,1R,70120,148280,B10,OC,0000,21105501
02,1039,PO,70340,149400,I10,B10,OC,500,21105501
03,1045,SI,70378,149025,I10,B07,PF,300,PN,17,BO,OI,21105501
"""

解决方案思路 #

观察数据可发现，字段错乱是因为不同**记录类型（第三列的值）**有不同的字段扩展规则。我们可以基于记录类型定义字段模板，将变长字段转换为结构化数据或补全为标准CSV格式。

方案1：转换为结构化字典列表

将每行数据转换为字典，保留字段语义，方便后续处理：

import csv

# 定义各记录类型的字段模板，可根据业务含义修改字段名
FIELD_TEMPLATES = {
    "PO": ["id", "code", "type", "val1", "val2", "cat1", "cat2", "status", "amount", "date"],
    "SI": ["id", "code", "type", "val1", "val2", "cat1", "cat2", "status", "amount", "param1", "num1", "param2", "param3", "date"],
    "SD": ["id", "code", "type", "val1", "val2", "cat1", "cat2", "status", "amount", "date"],
    "LD": ["id", "code", "type", "extra", "val1", "val2", "cat1", "cat2", "status", "amount", "date"],
    "LA": ["id", "code", "type", "extra", "val1", "val2", "cat2", "status", "amount", "date"]
}

def parse_legacy_csv(data_str):
    rows = data_str.strip().split("\n")
    parsed_data = []
    
    for row in rows:
        if not row:
            continue
        fields = row.split(",")
        record_type = fields[2]
        
        # 匹配对应模板，无匹配则用最长模板对齐
        template = FIELD_TEMPLATES.get(record_type, max(FIELD_TEMPLATES.values(), key=len))
        # 短行补空，长行截断（如需保留所有字段，可改为多余字段存入extra键）
        aligned_fields = fields[:len(template)] + [""] * (len(template) - len(fields))
        parsed_data.append(dict(zip(template, aligned_fields)))
    
    return parsed_data

# 执行解析
parsed_result = parse_legacy_csv(data)

# 验证结果（输出前3条）
for item in parsed_result[:3]:
    print(item)

说明：

• 模板字段名可根据实际业务含义调整，提升数据可读性
• 对于超长行（如示例中第38行），可扩展对应类型的模板字段，或在解析时将多余字段合并为一个extra字段统一存储

方案2：导出为标准对齐CSV

将结构化数据导出为标准CSV，所有行字段对齐，可直接用Excel、Pandas等工具处理：

def export_aligned_csv(parsed_data, output_path):
    # 收集所有出现过的字段，作为CSV表头
    all_fields = []
    seen_fields = set()
    for item in parsed_data:
        for key in item.keys():
            if key not in seen_fields:
                seen_fields.add(key)
                all_fields.append(key)
    
    # 写入CSV文件
    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=all_fields)
        writer.writeheader()
        writer.writerows(parsed_data)

# 导出对齐后的CSV
export_aligned_csv(parsed_result, "aligned_data.csv")

说明：

• 导出的CSV会自动为缺失字段填充空值，保证数据结构统一
• 表头包含所有出现过的字段，不会丢失任何原始数据

注意事项 #

• 若出现新的记录类型，只需在FIELD_TEMPLATES中添加对应的字段模板即可适配
• 若需处理超大CSV文件，可改为逐行读取解析，避免内存占用过高

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