如何使用Python将按TRANS/MTPNT/ASSET分组的多行TXT转换为CSV

你的数据属于固定行序的三元组结构：每组数据严格按照TRANS开头行→MTPNT开头行→ASSET开头行的顺序循环排列，总数据量仅数千行，计算负载极低，不需要用到重型分布式ETL框架。

根据技术栈不同可以选择两类方案，核心处理逻辑都是先按行分组再合并字段：

• 首选方案：Python 脚本处理
  灵活度最高，可自定义脏数据处理、字段转换规则，实现成本极低。核心处理步骤如下：

  1. 逐行读取源文件，跳过空行，每3行划为一个处理组
  2. 校验每组的行头标识是否符合TRANS/MTPNT/ASSET的顺序，避免数据乱序导致合并错位
  3. 按需提取三个行的字段拼接为单行，写入结果CSV即可
     参考实现代码：

  import csv
  
  INPUT_PATH = "你的源数据路径.txt"
  OUTPUT_PATH = "合并结果.csv"
  # 可自定义合并后的表头，按业务需要调整字段顺序
  MERGED_HEADERS = ["交易类型", "交易单号", "业务标识", "交易日期", "点位编号", "资产类型", "资产型号", "资产序列号"]
  
  with open(INPUT_PATH, "r", encoding="utf-8") as in_f, open(OUTPUT_PATH, "w", encoding="utf-8", newline="") as out_f:
      reader = csv.reader(in_f)
      writer = csv.writer(out_f)
      writer.writerow(MERGED_HEADERS)
      
      batch = []
      for row in reader:
          if not row:
              continue
          batch.append(row)
          # 凑齐3行就执行合并
          if len(batch) == 3:
              if batch[0][0] == "TRANS" and batch[1][0] == "MTPNT" and batch[2][0] == "ASSET":
                  # 此处可按需挑字段拼接，示例是直接拼接三个行的所有字段
                  merged_row = batch[0] + batch[1] + batch[2]
                  writer.writerow(merged_row)
              else:
                  print(f"异常分组，跳过：{batch}")
              batch = []
  

• 低代码方案：Power Query（Excel/Power BI 内置功能）
  适合无代码基础的用户，纯可视化操作即可完成处理：

  1. 把源数据导入Power Query，新增索引列，用Number.IntegerDivide([索引],3)生成分组ID
  2. 按分组ID做分组聚合，把每组的行内容聚合为列表
  3. 拆分列表为3列分别对应三类数据，展开所有字段后直接导出CSV即可

• 若存在缺行、乱序的脏数据，需先补充异常处理逻辑，把不符合规则的分组单独导出做人工校验，避免批量合并出错
• 如果需要做字段转换，比如把日期20211025转为2021-10-25格式、空值填充等，可在合并行的步骤中直接补充对应逻辑即可

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