
本文介绍一种健壮、可扩展的方法,使用原生 `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'Artikelnummer\d+:\s*(\S+)', row[col_idx])。
此方案彻底摆脱了对「单单元格内分号分割」的错误假设,直击问题本质——结构化数据的逻辑分组,是处理真实世界脏数据的工业级实践。










