Python：如何将多行结构化CSV解析为字典列表？

我来给你梳理一下具体的实现思路，分几个关键步骤来做，还会附上Python的示例代码，你可以参考：

1. 先拆分混排的原始文本 #

原始CSV内容是把H3、D1、D2、T1这些记录混在一起的，第一步要把它们拆成单独的记录行。可以用正则表达式匹配每个记录的起始标识（H3|、D1|、D2|、T1|），把整个文本拆分成一条条独立的记录。比如用re.split(r'(?=(H3|D1|D2|T1)\|)', raw_text)来拆分，这样就能得到每个单独的记录片段。

2. 遍历记录，构建嵌套结构 #

维护一个当前正在处理的文档对象（包含Header、Items、Trailer），以及当前正在处理的Item对象（包含D1字段和对应的D2子项列表），然后按顺序遍历每条拆分后的记录：

• 遇到H3开头的记录：初始化一个新的文档对象，解析H3的字段（去掉第一个标识位H3），把Header信息存进去，同时初始化Items为空列表。
• 遇到D1开头的记录：解析D1的字段，创建一个新的Item，把D1的字段存进去，同时给这个Item加一个sub_items列表用来存D2子项，然后把这个Item添加到当前文档的Items列表里，设置为当前活跃的Item。
• 遇到D2开头的记录：解析D2的字段，把它加到当前活跃Item的sub_items列表里。
• 遇到T1开头的记录：解析T1的字段，把Trailer信息存到当前文档里，然后把这个文档添加到最终的结果列表中，重置当前文档和活跃Item。

3. 字段类型转换（可选但推荐） #

原始数据都是字符串，根据需求把数字类型的字段（比如ID、数量、金额）转换成int或float，日期字段可以保留字符串或者转成datetime对象，让最终的数据结构更规范。

示例代码实现 #

import re
from typing import List, Dict, Tuple

def parse_structured_csv(raw_text: str) -> List[Dict]:
    # 拆分混排的记录，正则匹配开头的标识
    parts = re.split(r'(?=(H3|D1|D2|T1)\|)', raw_text.strip())
    # 把拆分后的片段组合成完整的记录行
    records = []
    for i in range(0, len(parts)-1, 2):
        records.append(f"{parts[i]}{parts[i+1]}")
    
    result = []
    current_doc = None
    current_item = None
    
    for record in records:
        fields = [f.strip() for f in record.split('|') if f.strip()]
        record_type = fields[0]
        
        if record_type == 'H3':
            # 解析Header字段
            header = (int(fields[1]), fields[2], int(fields[3]), fields[4])
            current_doc = {
                'Header': header,
                'Items': [],
                'Trailer': None
            }
        elif record_type == 'D1':
            # 解析D1主项字段
            d1_fields = (
                int(fields[1]), int(fields[2]), int(fields[3]), fields[4],
                fields[5], fields[6], float(fields[7]), float(fields[8]), fields[9]
            )
            current_item = {
                'main': d1_fields,
                'sub_items': []
            }
            current_doc['Items'].append(current_item)
        elif record_type == 'D2':
            # 解析D2子项字段
            d2_fields = (int(fields[2]), int(fields[3]), int(fields[4]))
            if current_item:
                current_item['sub_items'].append(d2_fields)
        elif record_type == 'T1':
            # 解析Trailer字段
            trailer = (int(fields[1]), int(fields[2]), float(fields[3]))
            if current_doc:
                current_doc['Trailer'] = trailer
                result.append(current_doc)
                current_doc = None
                current_item = None
    
    return result

# 测试示例文本
raw_data = """H3|509596|OUT|1653128|06/11/2018| D1|1653128|1|390|MXT586|EA|EA|55.600|219.99|Product 1 D2|1653128|1|900|390| T1|1653128|999|1000.000| H3|509597|OUT|1653128|06/11/2018| D1|1653128|1|390|MXT586|EA|EA|55.600|219.99|Product 2 D2|1653128|1|900|390| D2|1653128|2|600|430| T1|1653128|999|2164.000|"""

parsed_result = parse_structured_csv(raw_data)
print(parsed_result)

这个代码会把你提供的示例文本转换成目标嵌套结构，每个文档是一个字典，包含Header（元组）、Items（列表，每个Item包含主项字段和子项列表）、Trailer（元组）。如果需要调整字段顺序、类型或者结构细节，直接修改对应的字段解析部分即可。

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