如何实现从文件夹或Zip包中通用获取文件的流式方案？

问题描述 #

给定以下Python代码，需要实现一种通用的文件获取方式，将目录和Zip包统一包装为流式I/O对象，实现在线读取其中的文件，要求采用高效字节流（不加载整个源到内存）且流支持seek操作，可作为其他解码库的输入：

from pathlib import Path
import re
import zipfile as zip

def check_if_zip(path: Path) -> bool:
    return path.is_file() and (
        re.search(r"\.zip$", path.name, re.IGNORECASE) is not None
    )

sources: list[Path] = [
    Path("Foo"),  # 目录
    Path("Bar.zip"),  # Zip包
]

for source in sources:
    source_is_zip = check_if_zip(source)

    # 是否可以将每个源统一包装为某种流式I/O对象或类，
    # 实现在线读取文件夹或Zip包中的文件？需要一种与存储介质无关的文件获取方式，
    # 要求采用高效字节流（不将整个源加载到内存），且流支持seek操作，
    # 可作为其他库解码文件的输入。

解决方案 #

我们可以通过定义统一的抽象接口，为目录和Zip包分别实现适配器类，从而提供介质无关的API。

1. 定义抽象基类 #

先规定必须实现的核心方法：获取文件列表、打开文件并返回可seek的字节流。

from abc import ABC, abstractmethod
from pathlib import Path
import io
import zipfile as zip

class FileSource(ABC):
    @abstractmethod
    def list_files(self) -> list[str]:
        """返回源中的所有文件相对路径"""
        pass

    @abstractmethod
    def open_file(self, relative_path: str) -> io.IOBase:
        """打开指定文件，返回可seek的字节流"""
        pass

2. 实现目录适配器 #

针对本地目录，直接使用标准文件操作，返回的文件对象天然支持流式读取和seek。

class DirectorySource(FileSource):
    def __init__(self, dir_path: Path):
        self.dir_path = dir_path

    def list_files(self) -> list[str]:
        # 递归获取所有文件的相对路径
        return [str(path.relative_to(self.dir_path)) for path in self.dir_path.rglob("*") if path.is_file()]

    def open_file(self, relative_path: str) -> io.IOBase:
        file_path = self.dir_path / relative_path
        # 二进制只读模式打开，返回的BufferedReader支持seek和流式读取
        return open(file_path, "rb")

3. 实现Zip包适配器 #

利用zipfile.ZipFile.open()方法，返回的ZipExtFile对象支持seek操作，且不会加载整个Zip包到内存。

class ZipSource(FileSource):
    def __init__(self, zip_path: Path):
        self.zip_path = zip_path
        self.zip_file = None  # 延迟打开，减少资源占用

    def _ensure_zip_open(self):
        if self.zip_file is None:
            self.zip_file = zip.ZipFile(self.zip_path, "r")

    def list_files(self) -> list[str]:
        self._ensure_zip_open()
        # 过滤掉Zip中的目录
        return [info.filename for info in self.zip_file.infolist() if not info.is_dir()]

    def open_file(self, relative_path: str) -> io.IOBase:
        self._ensure_zip_open()
        # 返回的ZipExtFile支持seek和流式读取
        return self.zip_file.open(relative_path, "r")

4. 工厂函数统一创建源 #

通过工厂函数自动判断源类型，返回对应的适配器实例。

def create_file_source(path: Path) -> FileSource:
    if path.is_file() and path.suffix.lower() == ".zip":
        return ZipSource(path)
    elif path.is_dir():
        return DirectorySource(path)
    else:
        raise ValueError(f"不支持的源类型: {path}")

5. 使用示例 #

现在可以用完全一致的逻辑处理目录和Zip包：

sources: list[Path] = [
    Path("Foo"),  # 目录
    Path("Bar.zip"),  # Zip包
]

for source_path in sources:
    source = create_file_source(source_path)
    print(f"源 {source_path} 中的文件:")
    for file_path in source.list_files():
        print(f"- {file_path}")
        # 演示seek和流式读取
        with source.open_file(file_path) as f:
            f.seek(10)  # 跳到第10个字节
            chunk = f.read(20)
            print(f"  从偏移量10读取的20字节: {chunk}")

核心特性 #

• 流式读取：仅在需要时读取数据，不会将整个文件或Zip包加载到内存。
• Seek支持：返回的流对象均实现了seek()和tell()方法，兼容大多数解码库的输入要求。
• 介质无关：调用方无需区分源类型，通过FileSource接口即可完成所有操作。

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