
本文介绍一种健壮、可扩展的方法,使用原生 `csv` 模块解析具有嵌套结构的 csv(如订单主表+明细行),将分散在多行中的字段(如 artikelnummer、styckpris)按逻辑分组并展开为 `artikelnummer_1`, `artikelnummer_2` 等标准化列,完美解决顺序混乱、数量不一、空主键行等实际难题。
在处理某些导出型 CSV(如电商订单报表)时,常见一种“主-子”混合格式:首行为订单头(含 Ordernummer),后续空 Ordernummer 的行则属于该订单的明细项(如多个 Artikelnummer 和对应价格)。这种结构无法直接用 pandas.read_csv() 解析——因为 pandas 默认按行列对齐,而此处逻辑关系是跨行的。
原提问者尝试用 .split(';') 处理单列字符串,但数据实际是物理多行、逻辑一对多,强行切分会导致错位、丢失或崩溃。正确思路是:逐行扫描,识别主记录(非空 Ordernummer)与子记录(空 Ordernummer),动态构建每条订单的完整字段映射。
以下为推荐实现方案,兼顾鲁棒性、可读性与可维护性:
✅ 核心策略
- 使用 Python 内置 csv 模块逐行读取(避免 pandas 对不规则结构的误解析)
- 维护一个 out_data 列表,每个元素为 dict,代表一条完整订单记录
- 遇到非空 Ordernummer → 新建记录,拷贝当前行所有字段(主表信息)
- 遇到空 Ordernummer → 向最后一条记录追加子字段(如 Artikelnummer_1, Artikelns styckpris inkl moms_1)
- 自动推断最大子项数,确保输出 CSV 列头完整覆盖所有可能列(如 _1, _2, _3)
? 完整可运行代码
import csv
import pandas as pd
def parse_nested_csv(
input_file: str = 'orderexport_new.csv',
output_file: str = 'ny_orderdata.csv',
main_key_col: int = 0, # Ordernummer 所在列索引(从0开始)
sub_cols: list = [1, 3], # 需要拆分为多列的子字段列索引(如 Artikelnummer, styckpris)
sub_col_names: list = ['Artikelnummer', 'Artikelns styckpris inkl moms']
):
"""
解析主-子混合CSV,将子行字段展开为带序号的列(如 Artikelnummer_1, Artikelnummer_2)
Args:
input_file: 输入CSV路径
output_file: 输出CSV路径
main_key_col: 主键列索引(Ordernummer所在列)
sub_cols: 子字段列索引列表(需展开的列)
sub_col_names: 对应子字段的基名列表(用于生成列名如 'Artikelnummer_1')
"""
out_data = []
max_sub_count = 0 # 记录全局最大子项数
with open(input_file, encoding='utf-8') as f:
reader = csv.reader(f)
# 跳过两行表头(根据你的示例:第一行为字段名,第二行为单位/说明)
try:
headers1 = next(reader) # 如 ['Ordernummer', 'Intakt inkl moms', ...]
headers2 = next(reader) # 如 ['', 'Artikelnummer', 'Antal', ...]
except StopIteration:
raise ValueError("CSV 文件至少需要两行表头")
for row in reader:
if len(row) <= main_key_col or not row[main_key_col].strip():
# 子行:空 Ordernummer → 追加到上一条记录
if not out_data:
continue # 忽略开头无主键的孤立子行
# 当前子项序号(从1开始)
current_sub_idx = len([k for k in out_data[-1].keys()
if k.startswith(sub_col_names[0] + '_')]) + 1
max_sub_count = max(max_sub_count, current_sub_idx)
# 为每个指定子列生成带序号的新键值
for i, col_idx in enumerate(sub_cols):
if col_idx < len(row):
key_name = f"{sub_col_names[i]}_{current_sub_idx}"
out_data[-1][key_name] = row[col_idx].strip()
else:
# 主行:新建记录,初始化主字段
new_record = {headers1[i]: row[i].strip() if i < len(row) else ""
for i in range(len(headers1))}
out_data.append(new_record)
# 构建最终列头:主字段 + 动态生成的子字段(最多 max_sub_count 个)
main_headers = [h for h in headers1 if h.strip()]
dynamic_headers = []
for name in sub_col_names:
for i in range(1, max_sub_count + 1):
dynamic_headers.append(f"{name}_{i}")
all_headers = main_headers + dynamic_headers
# 写入结果CSV
with open(output_file, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=all_headers)
writer.writeheader()
for record in out_data:
# 补全缺失的动态列(避免 KeyError)
padded_record = {k: record.get(k, "") for k in all_headers}
writer.writerow(padded_record)
print(f"✅ 成功解析 {len(out_data)} 条订单,最大子项数:{max_sub_count}")
print(f"? 输出列:{all_headers}")
return pd.DataFrame(out_data, columns=all_headers)
# --- 使用示例 ---
if __name__ == "__main__":
# 配置你的字段映射(关键!按实际CSV结构调整)
df_result = parse_nested_csv(
input_file='orderexport_new.csv',
output_file='ny_orderdata.csv',
main_key_col=0, # Ordernummer 在第0列
sub_cols=[1, 3], # "Intakt inkl moms"列(索引1)和"Dynamisk valutajusterad"列(索引3)需展开
sub_col_names=['Artikelnummer', 'Artikelns styckpris inkl moms']
)
print(df_result.head())⚠️ 关键注意事项
- 编码兼容性:示例中使用 utf-8,若原始文件为 latin1(如问题中所示),请将 encoding='utf-8' 改为 encoding='latin1'
- 表头处理:代码默认跳过前两行(适配你提供的表格结构)。若你的 CSV 只有一行表头,请删除 headers2 = next(reader) 并调整 headers1 逻辑
- 列索引校准:sub_cols 中的数字必须严格对应 CSV 的实际列索引(用 Excel 或文本编辑器确认)。例如 "Artikelnummer" 在第二行显示为第2列,但在数据行中可能是索引 1(0起始)
- 空值安全:自动填充缺失子字段为空字符串,避免 pandas 后续处理报错
- 扩展性:只需修改 sub_cols 和 sub_col_names 即可支持任意数量的子字段(如增加 Antal_1, Antal_2)
? 总结
此方法摒弃了对“单列内分号分割”的错误假设,直击问题本质——结构化解析跨行逻辑关系。它不依赖数据顺序,能自动适应不同订单的子项数量差异,并生成规整的宽表(wide format)供后续分析。对于类似 ERP/CRM 导出的不规则报表,这是最可靠、最易调试的解决方案。
立即学习“Python免费学习笔记(深入)”;










