动手学深度学习（Pytorch版）代码实践 -循环神经网络-52文本预处理

52文本预处理

"""
我们将解析文本的常见预处理步骤。 
这些步骤通常包括：
1.将文本作为字符串加载到内存中。
2.字符串拆分为词元（如单词和字符）。
3.建立一个词表，将拆分的词元映射到数字索引。
4.将文本转换为数字索引序列，方便模型操作。
"""
import collections
import re
from d2l import torch as d2l

# 读取数据集
# 从H.G.Well的时光机器中加载文本
# d2l.DATA_HUB['time_machine'] = (d2l.DATA_URL + 'timemachine.txt',
#                                 '090b5e7e70c295757f55df93cb0a180b9691891a')
# d2l.download('time_machine')

time_machine_dir = '../data/timemachine.txt'

def read_time_machine():
    """将时间机器数据加载到文本行的列表中"""
    with open(time_machine_dir, 'r') as f:
        lines = f.readlines()
    return [
        re.sub('[^A-Za-z]+', ' ', line).strip().lower() 
        for line in lines
    ]
"""
re.sub 是正则表达式模块 re 中的一个函数，用于替换字符串中的匹配项。
[^A-Za-z] 表示匹配任何不是大写字母（A-Z）或小写字母（a-z）的字符。
+表示匹配一个或多个非字母字符
' ' 是替换字符串，表示将所有匹配的非字母字符替换为空格。

strip() 方法用于移除字符串首尾的空白字符。
lower() 方法将字符串中的所有字符转换为小写。
"""

lines = read_time_machine()
# print(f'# 文本总行数: {len(lines)}')
# print(lines[0])
# print(lines[10])
# # 文本总行数: 3221
# the time machine by h g wells
# twinkled and his usually pale face was flushed and animated the


# 词元化
def tokenize(lines, token='word'):  #@save
    """将文本行拆分为单词或字符词元"""
    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)
# print(tokens[0])
# ['the', 'time', 'machine', 'by', 'h', 'g', 'wells']

# 词表
# 构建一个字典，通常也叫做词表（vocabulary）
# 用来将字符串类型的词元映射到从0开始的数字索引中
class Vocab:
    """文本词表"""
    def __init__(self, tokens=None, min_freq=0, reserved_tokens=None) -> None:
        if tokens is None:
            tokens = []
        if reserved_tokens is None:
            reserved_tokens = []
        
        # 词元频率
        counter = count_corpus(tokens)
        # print(counter)
        # Counter({'the': 1, 'time': 1, 'machine': 1, 'by': 1, 'h': 1, 'g': 1, 'wells': 1})

        # 按出现频率排序
        self._token_freqs = sorted(counter.items(), key=lambda x: x[1],
                                   reverse=True)
        # 未知词元的索引为0
        self.idx_to_token = ['<unk>'] + reserved_tokens
        """
        这里，self.idx_to_token 是一个列表，包含所有词元及其对应的索引。
        '<unk>' 是一个特殊的词元，表示未知词元（unknown token）。
        reserved_tokens 是一个列表，包含预留的词元，可以是其他特殊词元，
        如 '<pad>'、'<bos>'（句子开始）和 '<eos>'（句子结束）。
        ['<unk>'] + reserved_tokens 将 '<unk>' 添加到 reserved_tokens 的前面，
        创建一个包含未知词元和预留词元的列表。
        """
        self.token_to_idx = {token: idx
                             for idx, token in enumerate(self.idx_to_token)}
        # enumerate(self.idx_to_token) 会生成一个迭代器，
        # 产生一系列的 (idx, token) 元组，其中 idx 是索引，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) - 1

    def __len__(self):
        return len(self.idx_to_token)

    def __getitem__(self, tokens):
        if not isinstance(tokens, (list, tuple)): # 检查 tokens 是否是列表或元组
            # 尝试获取词元 tokens 对应的索引。
            # 如果词元不在字典中，返回 self.unk，表示未知词元的索引
            return self.token_to_idx.get(tokens, self.unk)
        return [self.__getitem__(token) for token in tokens]

    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
    def unk(self):  # 未知词元的索引为0
        return 0

    @property
    def token_freqs(self):
        return self._token_freqs

def count_corpus(tokens):  #@save
    """统计词元的频率"""
    # 检查 tokens 是否为空或是二维列表
    # len(tokens) == 0, 二维的tokens如果没有元素的话tokens[0]会报错，需要单独处理
    if len(tokens) == 0 or isinstance(tokens[0], list):
        # 将词元列表展平成一个列表
        # tokens = [token for line in tokens for token in line]
        flat_tokens = []
        for line in tokens:
            for token in line:
                flat_tokens.append(token)
        tokens = flat_tokens

    return collections.Counter(tokens) # 得到词元的频率

vocab = Vocab(tokens)
# print(list(vocab.token_to_idx.items())[:10])
# [('<unk>', 0), ('the', 1), ('i', 2), ('and', 3), ('of', 4),
#  ('a', 5), ('to', 6), ('was', 7), ('in', 8), ('that', 9)]


# 整合所有功能
"""
在使用上述函数时，我们将所有功能打包到load_corpus_time_machine函数中， 
该函数返回corpus（词元索引列表）和vocab（时光机器语料库的词表）。 
我们在这里所做的改变是：
1.为了简化后面章节中的训练，我们使用字符（而不是单词）实现文本词元化；
2.时光机器数据集中的每个文本行不一定是一个句子或一个段落，还可能是一个单词，
因此返回的corpus仅处理为单个列表，而不是使用多词元列表构成的一个列表。
"""
def load_corpus_time_machine(max_tokens=-1):  #@save
    """返回时光机器数据集的词元索引列表和词表"""
    lines = read_time_machine()
    tokens = tokenize(lines, 'char')
    vocab = Vocab(tokens)
    # 因为时光机器数据集中的每个文本行不一定是一个句子或一个段落，
    # 所以将所有文本行展平到一个列表中
    corpus = [vocab[token] for line in tokens for token in line]
    if max_tokens > 0:
        corpus = corpus[:max_tokens]
    return corpus, vocab

corpus, vocab = load_corpus_time_machine(10)
print(corpus)  # 输出: [对应的词元索引列表]
print(vocab.token_to_idx)  # 输出: 词元到索引的映射字典
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170