Python pandas如何动态收集有效DataFrame完成拼接

• 业务侧数据源为携带CSV附件的邮件，所有这类邮件统一存储在S3存储桶中
• 现有流程通过Python作业逐个读取CSV文件，处理生成标准化结构的DataFrame后，将所有DataFrame拼接为单个数据框供后续使用
• 实际运行时每批次收到的CSV文件数量不固定，且每个文件存在独立字段特性，必须为每个CSV单独做标准化处理生成DataFrame后再执行拼接，需要优化流程兼容动态数量的DataFrame输入

try:
   Read CSV
   Do stuff to prepare standard dataframe
except:
   print('fail')
...numerous try-catch blocks...
frames = [ALL CREATED DATAFRAMES]
result_frame = pandas.concat(frames)

需要实现自动收集所有实际成功创建的DataFrame对象，将其作为frames列表传入拼接逻辑。此前尝试使用dir()从生成器对象构建列表，但该方法仅能获取变量名的字符串，无法拿到对应的变量对象本身，需要可落地的实现方案。

核心思路非常简单：提前初始化一个空列表专门存有效DataFrame，每成功生成一个DataFrame就立刻追加到这个列表里，完全不需要靠反射方法捞全局变量。
不要写大量重复的零散try-catch块，把单文件的读取、处理逻辑封装成独立函数，遍历所有待处理CSV文件逐个执行即可，参考实现如下：

import pandas as pd
# 按你实际使用的S3 SDK导入对应模块，这里以常用的boto3为例
import boto3

s3_client = boto3.client('s3')
# 提前初始化存储有效DataFrame的空列表
valid_frames = []

# 替换为你实际获取当前批次待处理CSV文件列表的逻辑，返回值为每个CSV对应的S3存储信息
csv_file_list = [
    # 示例格式：{"bucket": "你的存储桶名", "key": "邮件附件路径/xxx.csv"}
]

def process_single_csv(s3_file_info: dict) -> pd.DataFrame | None:
    """处理单个S3存储的CSV文件，处理成功返回标准化DataFrame，失败返回None"""
    try:
        # 读取S3上的CSV文件，按你实际的读取逻辑调整
        s3_obj = s3_client.get_object(Bucket=s3_file_info["bucket"], Key=s3_file_info["key"])
        df = pd.read_csv(s3_obj["Body"])
        # 此处插入你原有的单文件标准化处理逻辑
        # Do stuff to prepare standard dataframe
        return df
    except Exception as e:
        print(f'处理文件{s3_file_info["key"]}失败，错误信息：{str(e)}')
        return None

# 遍历所有待处理CSV，逐个执行处理
for file_info in csv_file_list:
    processed_df = process_single_csv(file_info)
    # 仅将处理成功的DataFrame加入有效列表
    if processed_df is not None:
        valid_frames.append(processed_df)

# 拼接前增加非空判断，避免全量文件处理失败时concat空列表报错
if valid_frames:
    result_frame = pd.concat(valid_frames, ignore_index=True)
else:
    # 无有效数据时的处理逻辑按业务需求自定义，比如返回空结构表
    result_frame = pd.DataFrame()

方案说明 #

• 该方案完全不需要使用dir()这类反射方法获取变量，所有有效DataFrame在生成时就被纳入统一管理，不会混入无关变量，也不会遗漏成功生成的DataFrame
• 将零散的try-catch逻辑收拢到单文件处理函数中，后续调整单文件处理规则只需要修改一处代码，可维护性远高于堆N个重复try-catch块的写法
• 新增的空列表判断可以避免某批次所有CSV都处理失败时，pd.concat接收空列表直接抛出异常的问题
• 如果你因为特殊原因必须保留原有零散写try-catch的代码结构，只需要提前定义好valid_frames = []，在每个try块中确认DataFrame生成成功后，执行valid_frames.append(你的DataFrame变量名)即可，最终直接拼接这个列表就能达到同样效果，只是这种写法冗余度高，不推荐使用。

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