Python无需解压读取嵌套7z文件，将CSV存入Pandas DataFrame

目录结构 #

我有如下嵌套结构的压缩目录：

folder_to_zip
    - file_1.csv
    - folder_1.zip
        1. file_2.csv
        2. file_3.csv
        3. folder_2.zip
            **.**file_4.csv
            **.**file_5.csv
            **.** file_6.csv
    -file_7.csv

需求 #

将其中每个CSV文件分别存入不同的Pandas DataFrame，避免解压后占用过多存储空间（原压缩包仅639MB，解压后达7.66GB）。

现有代码问题 #

尝试编写的代码逻辑完全错误，无法实现需求，代码如下：

from py7zr import SevenZipFile as szf
import os
import pandas as pd


def unzip_(folder_to_zip):
    dfs= []
    if not folder_to_zip.endswith('.csv'):
        dfs.append(pd.read_csv(folder_to_zip))
    else:      
        with szf(folder_to_zip, 'r') as z:
            for f in z.getnames():
                dfs += unzip_(f)
    return dfs       

问题分析与修正方案 #

核心问题点 #

1. 逻辑颠倒：把非CSV文件当成CSV读取，反而试图解析CSV文件为压缩包
2. 嵌套压缩包处理错误：直接传递内部文件名无法访问，需要从父压缩包中读取文件流
3. 格式不匹配：目录中是.zip格式，但代码用py7zr（仅处理7z格式），未适配zip格式

修正后的代码实现 #

使用zipfile处理zip压缩包，递归遍历并直接读取内部CSV的文件流到Pandas，全程在内存中操作，无需解压到磁盘：

import zipfile
import pandas as pd
from io import BytesIO

def extract_csv_to_dfs(archive_path):
    dfs = []
    with zipfile.ZipFile(archive_path, 'r') as zf:
        for file_name in zf.namelist():
            # 跳过目录项
            if file_name.endswith('/'):
                continue
            # 读取CSV文件到DataFrame
            if file_name.lower().endswith('.csv'):
                with zf.open(file_name) as f:
                    df = pd.read_csv(f)
                    dfs.append(df)
            # 递归处理嵌套的zip压缩包
            elif file_name.lower().endswith('.zip'):
                with zf.open(file_name) as nested_zip:
                    # 用BytesIO包装字节流，避免写入本地磁盘
                    nested_zip_io = BytesIO(nested_zip.read())
                    dfs += extract_csv_to_dfs(nested_zip_io)
    return dfs

# 调用示例
if __name__ == "__main__":
    all_dfs = extract_csv_to_dfs("folder_to_zip.zip")
    print(f"共读取到 {len(all_dfs)} 个CSV文件")

进阶优化（区分不同CSV文件） #

如果需要明确每个DataFrame对应的文件路径，可以返回字典格式：

import zipfile
import pandas as pd
from io import BytesIO

def extract_csv_to_dfs(archive_path, parent_prefix=""):
    df_map = {}
    with zipfile.ZipFile(archive_path, 'r') as zf:
        for file_name in zf.namelist():
            if file_name.endswith('/'):
                continue
            # 构建完整的嵌套路径
            full_path = f"{parent_prefix}/{file_name}" if parent_prefix else file_name
            if file_name.lower().endswith('.csv'):
                with zf.open(file_name) as f:
                    df_map[full_path] = pd.read_csv(f)
            elif file_name.lower().endswith('.zip'):
                with zf.open(file_name) as nested_zip:
                    nested_zip_io = BytesIO(nested_zip.read())
                    nested_map = extract_csv_to_dfs(nested_zip_io, full_path)
                    df_map.update(nested_map)
    return df_map

# 调用示例
if __name__ == "__main__":
    csv_df_map = extract_csv_to_dfs("folder_to_zip.zip")
    print(f"共读取到 {len(csv_df_map)} 个CSV文件")
    # 遍历查看每个文件对应的DataFrame
    for file_path, df in csv_df_map.items():
        print(f"文件 {file_path} 的行数：{len(df)}")

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