如何以流处理方式分割含多格式文档的分隔符文件并使用Python解析器处理？

嘿，这个场景我太有共鸣了！处理那种用---（换行包裹的三个短横线）分隔的多格式混合文档，一次性把整个文件读进内存拆分的方法，遇到大组件时确实会内存爆炸，流式处理才是更靠谱的选择。

先回顾下你提到的初始方案——虽然简单但有明显的内存瓶颈：

import pathlib, io, yaml, csv, json
partone, parttwo, partthree = pathlib.Path("file").read_text().split("\n---\n")
print(yaml.load(io.StringIO(partone)))
print(tuple(csv.reader(io.StringIO(parttwo))))
print(json.load(io.StringIO(partthree)))

这种方法小文件跑起来没问题，但要是某个组件是几百MB的CSV，内存直接就扛不住了。所以我们需要一种能边读边拆、遇到分隔符就暂停的流式方案。

你设想的两种理想用法真的很友好：要么返回一个文件对象迭代器，要么是一个能把分隔符当成“临时EOF”的包装文件对象。自己实现后者确实要处理不少边缘情况（比如read/readline带size参数时，分隔符被拆成两半读取的情况），不过这里有几个可行的方案分享给你：

方案一：自定义IO包装器（贴合你想要的“伪EOF”体验） #

我们可以继承Python的io.TextIOBase基类，实现一个包装器，它会在读取时跟踪分隔符，遇到时就停止当前部分的读取，下次调用读取方法时自动跳过分隔符，继续下一部分。下面是一个简化的文本模式实现：

import io

class SplitTextIO(io.TextIOBase):
    def __init__(self, file, separator):
        self.file = file
        self.sep = separator
        self.sep_len = len(separator)
        self.buffer = []
        self.finished_part = False
        self.eof = False

    def read(self, size=None):
        if self.finished_part:
            # 重置状态，跳过分隔符
            self.finished_part = False
            buffer_str = ''.join(self.buffer)
            if buffer_str.startswith(self.sep):
                self.buffer = [buffer_str[self.sep_len:]]
            else:
                # 从文件读取直到跳过完整分隔符
                while True:
                    chunk = self.file.read(self.sep_len)
                    if not chunk:
                        self.eof = True
                        break
                    buffer_str += chunk
                    if buffer_str.endswith(self.sep):
                        self.buffer = [buffer_str[:-self.sep_len]]
                        break
                    elif len(buffer_str) > self.sep_len:
                        buffer_str = buffer_str[1:]

        if self.eof:
            return ''

        result = []
        remaining = size if size is not None else float('inf')

        while remaining > 0 and not self.finished_part and not self.eof:
            chunk = self.buffer.pop(0) if self.buffer else self.file.read(1024)
            if not chunk:
                self.eof = True
                break

            sep_pos = chunk.find(self.sep)
            if sep_pos != -1:
                # 截取分隔符前的内容，剩余部分放回缓冲区
                result.append(chunk[:sep_pos])
                remaining -= sep_pos
                self.buffer.append(chunk[sep_pos+self.sep_len:])
                self.finished_part = True
            else:
                take = min(remaining, len(chunk))
                result.append(chunk[:take])
                remaining -= take
                if take < len(chunk):
                    self.buffer.append(chunk[take:])

        return ''.join(result)

    def readline(self, size=None):
        if self.finished_part:
            self.finished_part = False
            # 跳过分隔符逻辑同read方法
            buffer_str = ''.join(self.buffer)
            if buffer_str.startswith(self.sep):
                self.buffer = [buffer_str[self.sep_len:]]
            else:
                while True:
                    chunk = self.file.readline()
                    if not chunk:
                        self.eof = True
                        break
                    buffer_str += chunk
                    if buffer_str.endswith(self.sep):
                        self.buffer = [buffer_str[:-self.sep_len]]
                        break

        if self.eof:
            return ''

        line_buffer = []
        remaining = size if size is not None else float('inf')

        while remaining > 0 and not self.finished_part and not self.eof:
            chunk = self.buffer.pop(0) if self.buffer else self.file.readline()
            if not chunk:
                self.eof = True
                break

            sep_pos = chunk.find(self.sep)
            if sep_pos != -1:
                line_buffer.append(chunk[:sep_pos])
                remaining -= len(chunk[:sep_pos])
                self.buffer.append(chunk[sep_pos+self.sep_len:])
                self.finished_part = True
                break
            else:
                take = min(remaining, len(chunk))
                line_buffer.append(chunk[:take])
                remaining -= take
                if take < len(chunk):
                    self.buffer.append(chunk[take:])
                    break

        return ''.join(line_buffer)

    def close(self):
        self.file.close()

# 封装成你想要的splitfile函数
def splitfile(file, separator="\n---\n"):
    return SplitTextIO(file, separator)

# 测试用法，和你设想的完全一致！
with open("test_file.txt", "r") as f:
    with splitfile(f) as sf:
        import yaml, csv, json
        print(yaml.load(sf, Loader=yaml.FullLoader))  # 处理第一部分YAML
        print(tuple(csv.reader(sf)))                  # 处理第二部分CSV
        print(json.load(sf))                          # 处理第三部分JSON

这个实现处理了文本模式下的read和readline方法，完美贴合你想要的“一个文件对象多次复用，分隔符当临时EOF”的需求。你可以根据实际情况扩展（比如支持二进制模式、调整读取块大小等）。

方案二：流式拆分生成器（更轻量的折中方案） #

如果你觉得自定义IO类太繁琐，也可以用生成器实现流式拆分，把每个组件的内容逐个生成出来，再包装成StringIO传给解析器：

import itertools
import io

def split_stream(file, separator):
    sep_len = len(separator)
    buffer = []
    # 按块读取文件，直到EOF
    for chunk in iter(lambda: file.read(1024), ''):
        buffer.append(chunk)
        buffer_str = ''.join(buffer)
        while True:
            sep_pos = buffer_str.find(separator)
            if sep_pos == -1:
                # 保留可能的部分分隔符在缓冲区，避免漏匹配
                if len(buffer_str) > sep_len:
                    buffer = [buffer_str[-(sep_len-1):]]
                else:
                    buffer = [buffer_str]
                break
            # 输出当前组件内容
            yield buffer_str[:sep_pos]
            # 跳过分隔符，继续处理剩余内容
            buffer_str = buffer_str[sep_pos+sep_len:]
            buffer = [buffer_str]
    # 输出最后一个组件
    if buffer:
        yield ''.join(buffer)

# 用法示例
with open("file", "r") as f:
    components = (io.StringIO(part) for part in split_stream(f, "\n---\n"))
    import yaml, csv, json
    print(yaml.load(next(components), Loader=yaml.FullLoader))
    print(tuple(csv.reader(next(components))))
    print(json.load(next(components)))

这个方案代码更简洁，虽然需要手动把每个组件包装成StringIO，但也实现了流式处理，不会一次性加载整个文件到内存。

总结 #

如果你追求最贴合你设想的接口体验，自定义TextIOBase包装器是最佳选择；如果想快速实现功能，流式拆分生成器是更轻量的方案。两种方法都能解决大文件内存占用过高的问题。

备注：内容来源于stack exchange，提问作者XZS