如何用 Python 将 CSV 中的嵌套子行(分号/换行结构)智能拆分为多列

4次阅读

如何用 Python 将 CSV 中的嵌套子行(分号/换行结构)智能拆分为多列

本文介绍一种健壮、可扩展的方法,使用原生 `csv` 模块逐行解析非标准 csv(含空行关联、多级子记录),将如 `”artikelnummer1: 27404475; artikelnummer2: 75109997″` 类型的混合字段,按逻辑关系准确展开为 `artikelnummer_1`、`artikelnummer_2` 等独立列,并兼容不规则数据顺序与缺失项。

在实际业务导出的 CSV 文件中(如电商订单系统),常遇到「非平面化」结构:主记录(如 Ordernummer)与子记录(如多个 Artikelnummer 及其对应价格)交错排列,且子记录行首为空。这种格式无法直接用 pandas.read_csv() 解析为规整二维表——因为 pandas 默认按行列对齐,而此处的「逻辑行」跨越物理多行。

原始代码尝试用 str.split(‘;’) 处理单列字符串,但存在根本性缺陷:

  • ❌ 假设所有子字段都存在于同一单元格内(实际是跨行分布);
  • ❌ 忽略了 Ordernummer 为空行时的上下文继承逻辑;
  • ❌ 字典初始化错误(Intakt_inkl_moms = {} 却调用 .append());
  • ❌ 未处理字段名中的冒号、空格及多值映射关系。

✅ 正确解法是按物理行流式解析,识别「主行」(Ordernummer 非空)与「子行」(首列为空),动态构建每个订单的完整属性字典,并自动推导最大子项数量以生成完整列头。

以下为生产就绪的解决方案:

立即学习Python免费学习笔记(深入)”;

import csv import pandas as pd  def parse_nested_csv(     input_path: str,     output_path: str,     main_key_col: int = 0,  # Ordernummer 所在列索引(默认第0列)     sub_cols: list[int] = [1, 3],  # 需要拆分为多列的子字段列索引,如 [Artikelnummer, Styckpris]     sub_names: list[str] = ['Artikelnummer', 'Artikelns_styckpris_inkl_moms'] ):     """     解析含嵌套子行的CSV:主行定义订单,空首列行补充子项。     支持任意数量子列,自动命名如 'Artikelnummer_1', 'Artikelnummer_2'...     """     records = []     current_record = None     max_sub_count = 0      with open(input_path, encoding='utf-8') as f:         reader = csv.reader(f)          # 跳过两行表头(根据你的示例)         try:             next(reader)  # 第一行表头             next(reader)  # 第二行表头(单位/说明行)         except StopIteration:             pass  # 若无双表头则忽略          for row in reader:             # 安全处理空行或短行             if not row or len(row) <= main_key_col:                 continue              # 判断是否为主行:主键列非空             if row[main_key_col].strip():                 # 保存上一个记录(如有)                 if current_record is not None:                     records.append(current_record)                 # 新建主记录:用第一行表头作为键(需预先读取)                 # 这里简化:假设第一行表头已知,或动态捕获                 current_record = {'Ordernummer': row[main_key_col].strip()}                 # 复制其他主列值(如 Intakt inkl moms)                 for i, val in enumerate(row):                     if i != main_key_col and i < len(sub_cols):                         # 主行中非子列字段直接保留                         pass                 # 初始化子项计数器                 sub_index = 0             else:                 # 子行:仅当有主记录时才处理                 if current_record is None:                     continue                  sub_index += 1                 max_sub_count = max(max_sub_count, sub_index)                  # 为每个指定子列生成带序号的新字段                 for col_idx, base_name in zip(sub_cols, sub_names):                     if col_idx < len(row):                         value = row[col_idx].strip()                         key = f'{base_name}_{sub_index}'                         current_record[key] = value          # 添加最后一个记录         if current_record is not None:             records.append(current_record)      # 生成完整列名(主列 + 动态子列)     base_columns = ['Ordernummer']  # 可扩展为主表头列表     dynamic_columns = []     for name in sub_names:         for i in range(1, max_sub_count + 1):             dynamic_columns.append(f'{name}_{i}')      all_columns = base_columns + dynamic_columns      # 写入新CSV     with open(output_path, 'w', newline='', encoding='utf-8') as f:         writer = csv.DictWriter(f, fieldnames=all_columns)         writer.writeheader()         for record in records:             # 补全缺失的动态列(避免 KeyError)             padded_record = {k: record.get(k, '') for k in all_columns}             writer.writerow(padded_record)      # 返回 DataFrame 便于后续分析     return pd.DataFrame(records, columns=all_columns)  # 使用示例 if __name__ == '__main__':     df = parse_nested_csv(         input_path='orderexport_new.csv',         output_path='ny_orderdata.csv',         main_key_col=0,         sub_cols=[1, 3],  # 对应 "Intakt inkl moms" 和 "Artikelns styckpris inkl moms" 列         sub_names=['Artikelnummer', 'Artikelns_styckpris_inkl_moms']     )     print("✅ 解析完成!生成列:", df.columns.tolist())     print(df.head())

关键优势与注意事项:

  • 鲁棒性优先:显式跳过空行、短行;用 strip() 清理空白;动态推导最大子项数,避免硬编码列数。
  • 灵活配置:通过 sub_cols 和 sub_names 参数,可轻松适配新增子字段(如增加 Antal 列)。
  • 列名规范:自动生成 Artikelnummer_1, Artikelnummer_2 等语义化名称,便于后续 pandas 分组或透视。
  • 编码安全:显式指定 utf-8 编码(若原始文件为 latin1,请改为 encoding='latin1')。
  • 内存友好:流式处理,不加载全量数据到内存,适合大文件。

? 提示:若你的 CSV 表头结构不同(如无双表头、主键列非第0列),只需调整 main_key_col 和跳过表头的逻辑即可。对于更复杂的键值对解析(如 "Artikelnummer1: 27404475"),可在子行处理中加入正则提取:re.search(r'Artikelnummerd+:s*(S+)', row[col_idx])。

此方案彻底摆脱了对「单单元格内分号分割」的错误假设,直击问题本质——结构化数据的逻辑分组,是处理真实世界脏数据的工业级实践。

text=ZqhQzanResources