如何将含多表格的多个文本文件合并为Pandas DataFrame

一、直接处理原汇总文件（含多表格） #

核心思路是逐行读取文件，识别不同表格的分隔标识，将样本基础信息转为列，多行列的表格数据转为JSON字符串存储，最终每个样本对应DataFrame的一行。

1. 单个文件解析函数 #

import pandas as pd
import os
import glob
import json

def parse_summary_file(file_path):
    with open(file_path, 'r') as f:
        # 按行读取并拆分制表符，过滤空行
        lines = [line.strip('\n').split('\t') for line in f if line.strip()]
    
    sample_info = {}
    coverage_data = []
    variant_data = []
    interpretation_data = []
    
    current_section = None
    headers = None
    
    for line in lines:
        # 识别章节标题（以"Summary:"结尾的行）
        if line[0].endswith('Summary:'):
            current_section = line[0].replace(':', '')
            headers = None
            continue
        
        # 处理样本基础信息（键值对格式）
        if current_section == 'Sample Summary':
            if len(line) >=2 and line[0].endswith(':'):
                key = line[0].replace(':', '').strip()
                value = line[1].strip()
                sample_info[key] = value
        
        # 处理目标覆盖度表格
        elif current_section == 'Target Coverage Summary':
            if not headers:
                headers = [h.strip() for h in line if h.strip()]
            else:
                row = {headers[i]: line[i].strip() for i in range(len(headers))}
                coverage_data.append(row)
        
        # 处理变异信息表格
        elif current_section == 'Variant Summary':
            if not headers:
                headers = [h.strip() for h in line if h.strip()]
            else:
                row = {headers[i]: line[i].strip() for i in range(len(headers))}
                variant_data.append(row)
        
        # 处理药物解读表格
        elif current_section == 'Interpretations Summary':
            if not headers:
                headers = [h.strip() for h in line if h.strip()]
            else:
                row = {headers[i]: line[i].strip() for i in range(len(headers))}
                interpretation_data.append(row)
    
    # 将列表格式的表格数据转为JSON字符串，方便存入DataFrame
    sample_info['Target Coverage'] = json.dumps(coverage_data)
    sample_info['Variants'] = json.dumps(variant_data)
    sample_info['Interpretations'] = json.dumps(interpretation_data)
    
    # 返回单个样本的单行DataFrame
    return pd.DataFrame([sample_info])

2. 批量处理并合并所有文件 #

files_folder = "/data/TB/WA_dirty_prep_reports/"
all_sample_dfs = []

# 遍历所有文本文件
for file in glob.glob(os.path.join(files_folder, "*txt")):
    sample_df = parse_summary_file(file)
    all_sample_dfs.append(sample_df)

# 合并所有样本数据
final_df = pd.concat(all_sample_dfs, ignore_index=True)
print(final_df)

二、处理拆分后的单个表格文件 #

若已将每个样本的多个表格拆分为独立文件（如命名规则为{样本ID}_Sample.txt、{样本ID}_Coverage.txt等），可通过样本ID关联合并：

import pandas as pd
import os
import glob

files_folder = "/data/TB/WA_dirty_prep_reports/"

# 提取所有唯一样本ID（基于文件名前缀）
sample_ids = list(set([os.path.basename(f).split('_')[0] for f in glob.glob(os.path.join(files_folder, "*txt"))]))

all_sample_dfs = []

for sample_id in sample_ids:
    # 读取样本基础信息并转为字典
    sample_file = glob.glob(os.path.join(files_folder, f"{sample_id}_Sample.txt"))[0]
    sample_df = pd.read_csv(sample_file, sep='\t', header=None, names=['Key', 'Value'])
    sample_info = sample_df.set_index('Key')['Value'].to_dict()
    
    # 读取覆盖度数据并转为JSON
    coverage_file = glob.glob(os.path.join(files_folder, f"{sample_id}_Coverage.txt"))[0]
    coverage_df = pd.read_csv(coverage_file, sep='\t')
    sample_info['Target Coverage'] = coverage_df.to_json(orient='records')
    
    # 读取变异数据并转为JSON
    variant_file = glob.glob(os.path.join(files_folder, f"{sample_id}_Variant.txt"))[0]
    variant_df = pd.read_csv(variant_file, sep='\t')
    sample_info['Variants'] = variant_df.to_json(orient='records')
    
    # 读取解读数据并转为JSON
    interpretation_file = glob.glob(os.path.join(files_folder, f"{sample_id}_Interpretation.txt"))[0]
    interpretation_df = pd.read_csv(interpretation_file, sep='\t')
    sample_info['Interpretations'] = interpretation_df.to_json(orient='records')
    
    all_sample_dfs.append(pd.DataFrame([sample_info]))

final_df = pd.concat(all_sample_dfs, ignore_index=True)
print(final_df)

三、后续数据提取示例 #

如需展开JSON格式的列（如查看所有样本的变异详情）：

import json

# 展开变异信息列
variants_expanded = final_df.explode('Variants')
variants_expanded['Variants'] = variants_expanded['Variants'].apply(json.loads)
variants_df = pd.json_normalize(variants_expanded['Variants'])

# 合并回原样本基础信息
final_variants_df = pd.concat([variants_expanded.drop('Variants', axis=1), variants_df], axis=1)

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