如何在NLTK中从独立压缩包创建语料库（无需解压大文件）

你遇到的问题很典型——直接遍历目录只会拿到.zip文件，而NLTK的默认语料读取器不会自动解析压缩包内容。不用解压整个文件夹的话，我们可以用Python内置的zipfile模块直接读取每个压缩包里的文本文件，再把内容喂给NLTK处理。

下面是具体的解决思路和代码示例：

核心思路 #

• 遍历父目录下的所有.zip文件
• 对每个zip包，用zipfile.ZipFile打开，直接读取内部的.txt文件内容（不用解压到磁盘）
• 将读取到的文本内容整合到NLTK的语料处理流程中

代码实现 #

1. 遍历并读取单个zip包内的文本 #

首先写一个辅助函数，用来读取单个zip文件里的目标文本：

import zipfile
import os
from nltk.corpus.reader.plaintext import PlaintextCorpusReader

def read_zip_text(zip_path):
    # 打开zip文件（只读模式）
    with zipfile.ZipFile(zip_path, 'r') as zf:
        # 假设每个zip里只有一个txt文件，这里取第一个（或者根据文件名匹配）
        txt_filename = [f for f in zf.namelist() if f.endswith('.txt')][0]
        # 读取文件内容，注意编码可能需要调整，比如utf-8
        with zf.open(txt_filename) as txt_file:
            return txt_file.read().decode('utf-8')

2. 批量处理所有zip包并构建语料库 #

如果要构建可被NLTK直接使用的自定义语料库，我们可以自定义一个语料读取器，或者先收集所有文本内容再处理。这里提供两种方式：

方式一：生成器式批量读取（适合内存有限的场景）

这种方式不会一次性加载所有文本，而是逐个读取，适合处理20万个文件的大场景：

def corpus_generator(parent_dir):
    # 遍历父目录下的所有zip文件
    for filename in os.listdir(parent_dir):
        if filename.endswith('.zip'):
            zip_path = os.path.join(parent_dir, filename)
            try:
                text = read_zip_text(zip_path)
                yield text
            except Exception as e:
                print(f"处理{zip_path}时出错: {e}")
                continue

# 用法示例：遍历生成器获取文本，进行分词等处理
from nltk.tokenize import word_tokenize

for text in corpus_generator('Parent_dir'):
    tokens = word_tokenize(text)
    # 这里可以添加你的语料处理逻辑，比如统计、训练模型等

方式二：构建NLTK兼容的自定义语料库

如果你需要让这个语料库像NLTK自带的语料库一样使用（比如调用corpus.words()等方法），可以继承PlaintextCorpusReader并自定义读取逻辑：

from nltk.corpus.reader.api import CorpusReader
from nltk.corpus.reader.util import StreamBackedCorpusView, read_blankline_block

class ZipCorpusReader(CorpusReader):
    def __init__(self, root, fileids=r'.*\.zip', encoding='utf-8'):
        super().__init__(root, fileids, encoding)
    
    def _read_zip_block(self, stream):
        # 这里stream是zip文件的路径，我们需要读取内部txt内容
        zip_path = stream.name
        text = read_zip_text(zip_path)
        return [text]
    
    def raw(self, fileids=None):
        if fileids is None:
            fileids = self.fileids()
        return '\n'.join([read_zip_text(os.path.join(self.root, fid)) for fid in fileids])
    
    def words(self, fileids=None):
        return StreamBackedCorpusView(
            self.root, fileids, self._read_zip_block, word_tokenize, encoding=self.encoding
        )

# 初始化自定义语料库
corpus = ZipCorpusReader('Parent_dir')

# 用法示例：获取所有词
all_words = corpus.words()

注意事项 #

• 编码问题：如果你的文本文件不是utf-8编码，需要调整decode里的参数（比如gbk），可以先测试几个文件确定编码。
• 错误处理：20万个文件难免有损坏的zip，所以一定要加异常捕获，避免程序中途崩溃。
• 性能优化：如果处理速度慢，可以考虑用多进程/多线程批量处理，但要注意文件IO的并发限制。

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