如何用Python Pandas将CSV中每组3行内容合并提取为单行？

转换规则 #

• 识别Header列为Header的行作为系统（System），其下方至下一个Header行前的内容为该系统的组件
• 每个组件由连续3行组成，需将这3行合并为一行，对应新列规则：
  • 系统：所属Header行的Unit-Tag No.值
  • Unit Tag：组件第1行的Unit-Tag No.值
  • Doc Number：组件第3行的Unit-Tag No.值
  • Description：合并组件3行中Description instrument列的非空值，用; 分隔
  • Manufacturer：组件第1行的Manufacturer列值
  • Model Number：组件第2行的Manufacturer列值
  • EX Number：组件第3行的Manufacturer列值（非空则取）
  • Remarks：合并组件3行中Remarks列的非空值，用; 分隔
  • SE-TAG：组件第3行的Supplier列值

原CSV数据 #

Unit-Tag No.,Header,Description instrument,Manufacturer,Remarks,Supplier
CYG30EU005  Shaft vibration,Header,,,,
VSTTH,,Common alarm - Vibration,Vibro-Meter,hardwired (VM to PLC),Siemens Energy
BC000414P01E070,,Unit Control panel MMS,VM600,TEEES,
415012,,,DDD,,VSH9020
VSH,,Common alarm - Vibration,Siemens Energy,hardwired (VM to PLC),Siemens Energy
BC000414P01E070CIEIA,,Unit Control panel PLC,WinCC/PCS7,,
415013,,,,,VSH9020
CYG30EU010  Shaft position,Header,,,,
VSH,,Common alarm - Position,Vibro-Meter,hardwired (VM to PLC),Siemens Energy
1BC000414P01E070CIEIA,,Unit Control panel MMS,VM600,,
415013,,,,,VSH9021

目标格式 #

实现代码 #

import pandas as pd
import numpy as np

# 读取CSV，将空字符串转为NaN便于处理
df = pd.read_csv("your_file.csv", na_filter=False).replace("", np.nan)

# 1. 标记每行所属的系统
header_indices = df[df["Header"] == "Header"].index
df["System"] = np.nan
for idx in header_indices:
    system_name = df.loc[idx, "Unit-Tag No."]
    # 确定当前系统覆盖的行范围：当前Header行下一行到下一个Header行的上一行
    next_header_idx = header_indices[header_indices > idx].min() if any(header_indices > idx) else df.index.max() + 1
    df.loc[idx+1 : next_header_idx-1, "System"] = system_name

# 2. 过滤Header行，仅保留组件行
component_rows = df.drop(header_indices).dropna(subset=["System"]).reset_index(drop=True)

# 3. 按3行一组拆分，生成组件组编号
component_rows["Group"] = component_rows.index // 3

# 4. 对每个组件组应用转换规则
def process_component_group(group):
    if len(group) != 3:
        return pd.Series([np.nan]*9, index=["系统（System）", "单元标签（Unit Tag）", "文档编号（Doc Number）", "描述（Description）", "制造商（Manufacturer）", "型号（Model Number）", "防爆编号（EX Number）", "备注（Remarks）", "SE-TAG"])
    
    row1, row2, row3 = group.iloc[0], group.iloc[1], group.iloc[2]
    
    # 合并非空值，避免生成无效分隔符
    description = "; ".join(filter(pd.notna, [row1["Description instrument"], row2["Description instrument"], row3["Description instrument"]]))
    remarks = "; ".join(filter(pd.notna, [row1["Remarks"], row2["Remarks"], row3["Remarks"]]))
    
    return pd.Series({
        "系统（System）": row1["System"],
        "单元标签（Unit Tag）": row1["Unit-Tag No."],
        "文档编号（Doc Number）": row3["Unit-Tag No."],
        "描述（Description）": description if description else np.nan,
        "制造商（Manufacturer）": row1["Manufacturer"],
        "型号（Model Number）": row2["Manufacturer"],
        "防爆编号（EX Number）": row3["Manufacturer"] if pd.notna(row3["Manufacturer"]) else np.nan,
        "备注（Remarks）": remarks if remarks else np.nan,
        "SE-TAG": row3["Supplier"]
    })

# 执行转换并整理结果
result_df = component_rows.groupby(["System", "Group"]).apply(process_component_group).reset_index(drop=True)
result_df = result_df.fillna("")  # 将NaN转为空字符串匹配目标格式

# 输出或保存结果
print(result_df)
result_df.to_csv("structured_result.csv", index=False)

关键步骤说明 #

1. 系统分组标记：定位所有Header行，为后续组件行批量分配所属系统，解决跨系统归属问题
2. 组件分组拆分：通过整数除法生成组编号，将连续3行划分为一个组件单元
3. 非空值合并：用filter(pd.notna, ...)过滤空值后拼接，避免出现多余的分隔符
4. 异常处理：加入组长度检查，防止CSV末尾存在不完整组件行导致报错（可根据实际数据调整逻辑）

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