中文语料库构建过程详细教程

import re

re.compile('.*\.txt')

# 引入用到的库
from gensim.corpora.textcorpus import TextDirectoryCorpus
import locale
locale.setlocale(locale.LC_ALL, 'en_US.utf-8')
# 数据所在路径
dpath = r'D:\mysites\notebooks\data\test-corpus'


class MyTextDirCorpus(TextDirectoryCorpus):
    # 为了强制使用'utf8'编码, 我们复写了这个方法
    def getstream(self):
        """Yield documents from the underlying plain text collection (of one or more files).
        Each item yielded from this method will be considered a document by subsequent
        preprocessing methods.
        If `lines_are_documents` was set to True, items will be lines from files. Otherwise
        there will be one item per file, containing the entire contents of the file.
        """
        num_texts = 0
        for path in self.iter_filepaths():
            with open(path, 'rt', encoding='utf8') as f:
                if self.lines_are_documents:
                    for line in f:
                        yield line.strip()
                        num_texts += 1
                else:
                    content = f.read().strip()
                    yield content
                    num_texts += 1

        self.length = num_texts

# 实例化一个语料库, 
# 遍历的最小深度是1
# 设置lines_are_documents为False
# 后缀为txt的文件才会被当作是一个文档
corpus = MyTextDirCorpus(dpath, min_depth=1, pattern='.*\.txt',  lines_are_documents=False)
print('语料库计数:', len(corpus))
print('第一条:',next(corpus.get_texts()))

# 分词测试
from pyltp import Segmentor
new_doc = '实验室中测量无序响应中时间'
model_path = r'D:\mysites\text-characters\tcharacters\ltp\ltp_data\cws.model'
segmentor = Segmentor()  # 初始化实例
segmentor.load(model_path)
new_tokens = segmentor.segment(new_doc)
print('Tokens:', '\t'.join(new_tokens))
segmentor.release()  # 释放模型

class MyTextDirCorpus(TextDirectoryCorpus):

    def __init__(self, input, **kwargs):
        kwargs['tokenizer'] = self.tokenizer
        super().__init__(input, **kwargs)
    
    def tokenizer(self, text):
        if not hasattr(self, '_segmentor'):
            model_path = r'D:\mysites\text-characters\tcharacters\ltp\ltp_data\cws.model'
            segmentor = Segmentor()  # 初始化实例
            segmentor.load(model_path)
            self._segmentor = segmentor
        segmentor = self._segmentor
        return segmentor.segment(text)
    
    def __del__(self):
        '''释放资源'''
        if hasattr(self, '_segmentor'):
            self._segmentor.release()
        try:
            super().__del__()
        except AttributeError:
            pass
    
    # 为了强制使用'utf8'编码, 我们复写了这个方法
    def getstream(self):
        """Yield documents from the underlying plain text collection (of one or more files).
        Each item yielded from this method will be considered a document by subsequent
        preprocessing methods.
        If `lines_are_documents` was set to True, items will be lines from files. Otherwise
        there will be one item per file, containing the entire contents of the file.
        """
        num_texts = 0
        for path in self.iter_filepaths():
            with open(path, 'rt', encoding='utf8') as f:
                if self.lines_are_documents:
                    for line in f:
                        yield line.strip()
                        num_texts += 1
                else:
                    content = f.read().strip()
                    yield content
                    num_texts += 1

        self.length = num_texts

# 实例化一个语料库
# 注意我们传入了一个token_filters参数, 实际上是一个空列表, 意思是, 不要过滤词, 所有词都要
# 当然在有需要的情况下, 我们需要定义自己的token_filters
corpus = MyTextDirCorpus(dpath, min_depth=1, 
                         pattern='.*\.txt',  
                         lines_are_documents=False, 
                         tokenizer=tokenizer, 
                         token_filters=[])
# 保存词典到本地硬盘
dict_path = r'D:\mysites\notebooks\data\test-corpus\dictionary.model'
corpus.dictionary.save_as_text(fname=dict_path)
# 查看词典内容
corpus.dictionary.token2id

from gensim.corpora.dictionary import Dictionary

%%timeit
# 预加载字典
dic = Dictionary.load_from_text(dict_path)
MyTextDirCorpus(dpath, 
                dictionary=dic,
                min_depth=1, 
                 pattern='.*\.txt',  
                 lines_are_documents=False, 
                 tokenizer=tokenizer, 
                 token_filters=[])

%%timeit
# 每次生成词典
MyTextDirCorpus(dpath, 
                min_depth=1, 
                 pattern='.*\.txt',  
                 lines_are_documents=False, 
                 tokenizer=tokenizer, 
                 token_filters=[])

import pickle
from pathlib import Path

dic = Dictionary.load_from_text(dict_path)
corpus = MyTextDirCorpus(dpath, 
                dictionary=dic,
                min_depth=1, 
                 pattern='.*\.txt$',  
                 lines_are_documents=False, 
                 tokenizer=tokenizer, 
                 token_filters=[])

for fpath in corpus.iter_filepaths():
    fpath = Path(fpath)
    token_path = fpath.parent / (fpath.name + '.cached_tokens')
    txt = fpath.read_text(encoding='utf8').strip()
    tokens = corpus.tokenizer(txt)
    token_path.write_bytes(pickle.dumps(list(tokens)))
    
    

class MyTextDirCorpus(TextDirectoryCorpus):

    def __init__(self, input, **kwargs):
        kwargs['tokenizer'] = self.tokenizer
        super().__init__(input, **kwargs)
    
    def tokenizer(self, text):
        if not hasattr(self, '_segmentor'):
            model_path = r'D:\mysites\text-characters\tcharacters\ltp\ltp_data\cws.model'
            segmentor = Segmentor()  # 初始化实例
            segmentor.load(model_path)
            self._segmentor = segmentor
        segmentor = self._segmentor
        return segmentor.segment(text)
    
    def __del__(self):
        '''释放资源'''
        if hasattr(self, '_segmentor'):
            self._segmentor.release()
        try:
            super().__del__()
        except AttributeError:
            pass
    
    # 为了强制使用'utf8'编码, 我们复写了这个方法
    def getstream(self):
        """Yield documents from the underlying plain text collection (of one or more files).
        Each item yielded from this method will be considered a document by subsequent
        preprocessing methods.
        If `lines_are_documents` was set to True, items will be lines from files. Otherwise
        there will be one item per file, containing the entire contents of the file.
        """
        num_texts = 0
        for path in self.iter_filepaths():
            with open(path, 'rt', encoding='utf8') as f:
                if self.lines_are_documents:
                    for line in f:
                        yield line.strip()
                        num_texts += 1
                else:
                    content = f.read().strip()
                    yield content
                    num_texts += 1

        self.length = num_texts
        
    def get_texts_from_tokens(self):
        for fpath in self.iter_filepaths():
            fpath = Path(fpath)
            token_path = fpath.parent / (fpath.name + '.cached_tokens')
            yield pickle.loads(token_path.read_bytes())

%%timeit

dic = Dictionary.load_from_text(dict_path)
corpus = MyTextDirCorpus(dpath, 
                dictionary=dic,
                min_depth=1, 
                 pattern='.*\.txt$',  
                 lines_are_documents=False, 
                 tokenizer=tokenizer, 
                 token_filters=[])

for i in corpus.get_texts():
    pass

%%timeit

dic = Dictionary.load_from_text(dict_path)
corpus = MyTextDirCorpus(dpath, 
                dictionary=dic,
                min_depth=1, 
                 pattern='.*\.txt$',  
                 lines_are_documents=False, 
                 tokenizer=tokenizer, 
                 token_filters=[])

for i in corpus.get_texts_from_tokens():
    pass

# author: mlln.cn
# email: xxxspy@126.com
# qq: 675495787

class MyTextDirCorpus(TextDirectoryCorpus):

    def __init__(self, input, **kwargs):
        kwargs['tokenizer'] = self.tokenizer
        super().__init__(input, **kwargs)
    
    def tokenizer(self, text):
        if not hasattr(self, '_segmentor'):
            model_path = r'D:\mysites\text-characters\tcharacters\ltp\ltp_data\cws.model'
            segmentor = Segmentor()  # 初始化实例
            segmentor.load(model_path)
            self._segmentor = segmentor
        segmentor = self._segmentor
        return segmentor.segment(text)
    
    def __del__(self):
        '''释放资源'''
        if hasattr(self, '_segmentor'):
            self._segmentor.release()
        try:
            super().__del__()
        except AttributeError:
            pass
    
    # 为了强制使用'utf8'编码, 我们复写了这个方法
    def getstream(self):
        """Yield documents from the underlying plain text collection (of one or more files).
        Each item yielded from this method will be considered a document by subsequent
        preprocessing methods.
        If `lines_are_documents` was set to True, items will be lines from files. Otherwise
        there will be one item per file, containing the entire contents of the file.
        """
        num_texts = 0
        for path in self.iter_filepaths():
            with open(path, 'rt', encoding='utf8') as f:
                if self.lines_are_documents:
                    for line in f:
                        yield line.strip()
                        num_texts += 1
                else:
                    content = f.read().strip()
                    yield content
                    num_texts += 1

        self.length = num_texts
        
    def get_texts_from_tokens(self):
        for fpath in self.iter_filepaths():
            fpath = Path(fpath)
            token_path = fpath.parent / (fpath.name + '.cached_tokens')
            yield pickle.loads(token_path.read_bytes())
            
    def save_tokens(self):
        '''保存tokens到硬盘, 只需要运行一次'''
        for fpath in self.iter_filepaths():
            fpath = Path(fpath)
            token_path = fpath.parent / (fpath.name + '.cached_tokens')
            txt = fpath.read_text(encoding='utf8').strip()
            tokens = self.tokenizer(txt)
            token_path.write_bytes(pickle.dumps(list(tokens)))
    
    def save_dictionary(self, dpath):
        '''把字典保存到硬盘'''
        self.dictionary.save_as_text(fname=dpath)