循环神经网络-文本预处理

import collections

import re

from d2l import torch as d2l

d2l.DATA_HUB['time_machine'] = (d2l.DATA_URL + 'timemachine.txt','090b5e7e70c295757f55df93cb0a180b9691891a')

def read_time_machine():

    with open(d2l.download('time_machine'),'r') as f:

        lines = f.readlines()

    return [re.sub('[^A-Za-z]+',' ',line).strip().lower() for line in lines]

lines = read_time_machine()

print(f'文本总行数:{len(lines)}')

print(lines[0])

print(lines[10])

def tokenize(lines,token='word'):

    if token == 'word':

        return [line.split() for line in lines]

    elif token == 'char':

        return [list(line) for line in lines]

    else:

        print("错误: 未知的词元类型: " + token)

tokens = tokenize(lines)

for i in range(11):

    print(tokens[i])

class Vocab:

    def __init__(self, tokens=None, min_freq=0,reserved_tokens=None):

        if tokens is None:

            tokens = []

        if reserved_tokens is  None:

            reserved_tokens = []

        # 按照出现频率排序

        counter = count_corpus(tokens)

        """

        * counter.items() 是字典的一个方法。它返回一个包含字典所有项（(键, 值) 对，即 (key, value) tuples）的视图对象。例如，如果 counter 是 {'the': 10, 'a': 5}，counter.items() 会提供像 ('the', 10) 和 ('a', 5) 这样的元素。

        * sorted() 在对元素进行比较之前，会先将每个元素传递给 key 指定的函数，然后根据这些函数返回的结果进行排序。

        * lambda匿名函数见下

        """

        self._token_freqs = sorted(counter.items(), key=lambda x:x[1], reverse=True)

        # 未知词元的索引为 0

        self.idx_to_token = ['<unk>'] + reserved_tokens

        self.token_to_idx = {token: idx for idx, token in enumerate(self.idx_to_token)}

        for token, freq in self._token_freqs:

            if freq < min_freq:

                break

            if token not in self.token_to_idx:

                self.idx_to_token.append(token)

                self.token_to_idx[token] = len(self.idx_to_token)

    # 定义指定方法 `__len__`. 这个方法允许在这个类对象上使用内置函数 `len ()`。 (e.g., `len (my_vocab)`).

    def __len__(self):

        return len(self.idx_to_token)

     # 定义特殊方法 `__getitem__`。这允许使用方括号表示法进行查找（例如，`my_vocab ['hello']` 或 `my_voab [['hello'，'world']]）。

    def __getitem__(self, tokens):

        if not isinstance(tokens, (list, tuple)):

            return self.token_to_idx.get(tokens,self.unk)

        return [self.__getitem__(token) for token in tokens]

	# 进行反向查找 index--->token

    def to_tokens(self, indices):

        if not isinstance(indices, (list, tuple)):

            return self.idx_to_token[indices]

        return [self.idx_to_token[index]for index in indices]

      # `@property` 是一个装饰器，允许我们向调用属性的方式一样来调用函数。E.g., `my_vocab.unk` instead of `my_vocab.unk ()`.

    @property

    def unk(self):

        return 0

    @property

    def token_freqs(self):

        return self._token_freqs

def count_corpus(tokens):

    """统计词元出现的频率"""

    if len(tokens) == 0 or isinstance(tokens[0], list):

        # 将词元展开成一个列表

        tokens = [token for line in tokens for token in line]

    return collections.Counter(tokens)

vocab = Vocab(tokens)

print(list(vocab.token_to_idx.items())[:10])

for i in [0,10]:

    print('文本: ', tokens[i])

    print('索引: ', vocab[tokens[i]])

def load_corpus_time_machine(max_tokens=-1):

    """返回时光机器数据集的词元索引列表和词表"""

    lines = read_time_machine()

    tokens = tokenize(lines,'char')

    vocab = Vocab(tokens)

    # 因为时光机器数据集中的每个文本行不一定是一个句子或段落，所以将所有的文本行展平到一个列表中

    corpus = [vocab[token] for line in tokens for token in lines]

    if max_tokens > 0:

        corpus = corpus[:max_tokens]

    return corpus,vocab

corpus,vocab = load_corpus_time_machine()

len(corpus),len(vocab)