pandas导出CSV构造HTTP响应场景下如何拆分CSV文件

实现方案 #

HTTP单次响应无法直接返回多个独立CSV文件，拆分CSV后需要将所有分片打包为ZIP压缩包再返回，全流程可在内存中完成，无需落地本地临时文件，可直接适配你现有的pandas导出逻辑。 #

核心实现步骤

• 确定拆分规则（常用规则为按固定行数拆分、按指定字段值拆分）
• 在内存缓冲区中逐块生成拆分后的CSV分片
• 将所有分片写入内存中的ZIP压缩包
• 构造ZIP类型的HTTP下载响应返回

可直接复用的代码示例

首先引入依赖：

import urllib.parse
import io
import zipfile
import pandas as pd
# 根据你使用的web框架导入对应Response类，FastAPI/Flask/Django逻辑通用
from fastapi import Response

替换你原有CSV生成和响应构造的逻辑：

# 下载文件名改为zip后缀
_filename = "audit_log_export.zip"
attachment = 'attachment; filename*=utf-8\'\'' + urllib.parse.quote(_filename)

# 原有DataFrame构造逻辑保持不变
auditlog_data_frame = pd.DataFrame(data, columns=header)

# ---------------- CSV拆分逻辑 ----------------
# 配置拆分规则：示例为每10000行拆分为一个CSV，可按需调整chunk_size值
chunk_size = 10000
total_rows = len(auditlog_data_frame)

# 创建内存ZIP缓冲区
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_writer:
    # 逐片生成CSV写入压缩包
    for idx in range(0, total_rows, chunk_size):
        chunk_df = auditlog_data_frame.iloc[idx: idx + chunk_size]
        # 分片文件名可自定义命名规则
        chunk_file_name = f"audit_log_part_{idx//chunk_size + 1}.csv"
        # 内存中生成CSV内容
        csv_buffer = io.StringIO()
        # 建议使用utf-8-sig编码，避免Windows下Excel打开中文乱码
        chunk_df.to_csv(csv_buffer, index=False, encoding="utf-8-sig")
        # 写入压缩包
        zip_writer.writestr(chunk_file_name, csv_buffer.getvalue())

zip_buffer.seek(0)
# ---------------- 拆分逻辑结束 ----------------

# 构造响应返回，注意media_type改为zip对应类型
return Response(
    content=zip_buffer.getvalue(),
    media_type="application/zip",
    headers={"Content-Disposition": attachment}
)

自定义拆分规则

如果需要按指定字段拆分（比如按操作日期、操作类型分别生成独立CSV），只需要替换上述代码中的切片循环逻辑即可，示例：

# 示例：按operation_type字段值拆分，每个类型对应一个CSV
with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_writer:
    for field_val, group_df in auditlog_data_frame.groupby("operation_type"):
        chunk_file_name = f"audit_log_{field_val}.csv"
        csv_buffer = io.StringIO()
        group_df.to_csv(csv_buffer, index=False, encoding="utf-8-sig")
        zip_writer.writestr(chunk_file_name, csv_buffer.getvalue())

注意事项

• 全流程基于内存缓冲区实现，不会在服务器生成临时文件，无文件残留风险，性能优于本地文件读写方案
• 如果使用Flask框架，仅需将Response替换为flask.Response即可，其余逻辑无需改动
• 单分片大小建议控制在10万行以内，避免用户打开单个CSV时出现卡顿
• 若数据量超过百万级，建议将DataFrame构造改为分块读取逻辑，避免一次性占用过多内存

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