数据结构编程实践20讲(Python版)—19字典树

往期链接

19 字典树（Trie）

S1 说明

字典树，又称为前缀树，是一种树形数据结构，主要用于存储字符串集合，用于高效地完成字符串的插入和查找操作。树的核心思想是利用字符串的公共前缀 来减少存储空间和查找时间。

字典树结构

• 节点（Node）：每个节点代表一个字符串中的一个字符。
• 边（Edge）：从父节点到子节点的连接，表示字符的连接关系。
• 根节点（Root）：一个空节点，代表空字符串的开始。
• 结束标志（Terminal Flag）：用于标识某个节点是否为某个字符串的结尾。

字典树的构建与查找

• 构建（Insertion）
  • 从根节点开始。
  • 对于要插入的字符串，从第一个字符开始，逐个字符向下寻找对应的子节点。
  • 如果子节点存在，则移动到子节点；否则，创建新的子节点。
  • 重复上述步骤，直到处理完字符串的所有字符。
  • 标记当前节点为结束标志，表示一个完整字符串的结束。
• 查找（Search）
  • 从根节点开始。
  • 按照要查找的字符串，从第一个字符开始，逐个字符向下寻找对应的子节点。
  • 如果在某一步无法找到对应的子节点，则表示该字符串不在Trie中。
  • 如果成功找到所有字符对应的节点，并且最后的节点标记为结束标志，则表示Trie中存在该字符串。

字典树的特点

1. 优势

• 高效的字符串查询：Trie树可以在 O(m) 的时间内完成字符串的插入、删除和查找操作，其中 m 为字符串的长度。这与集合中元素的数量无关，避免了传统搜索算法中 O(log n) 或 O(n) 的时间复杂度。
• 前缀匹配：Trie天然支持前缀查询，可以方便地实现以某个前缀开头的所有字符串的检索。
• 节省存储空间：通过共享公共前缀，Trie可以减少存储重复的字符，尤其在字符串集合存在大量公共前缀的情况下。

2. 劣势

• 空间占用较大：对于字符集较大的情况（如UNICODE字符集），Trie的节点分支数会很大，可能导致空间浪费。
• 实现复杂度：相比于哈希表等数据结构，Trie的实现要复杂一些，维护起来也更为繁琐。
• 不支持部分操作：Trie不适合用于需要对字符串进行排序或范围查询的场景，因为它不维护元素的顺序关系。

字典树的应用领域

1. 字符串检索与自动补全

• 输入法：在输入法中，利用Trie可以高效地完成对用户输入前缀的匹配，提供候选词的自动补全。
• 搜索引擎：提供搜索关键词的实时提示功能，根据用户输入的前缀，快速返回可能的搜索词。

2. 词典存储与拼写检查

• 拼写检查器：将词典中的所有单词存储在Trie中，可以快速判断一个单词是否正确，以及提供可能的拼写建议。
• 敏感词过滤：利用Trie存储敏感词汇，在文本处理中快速检测并屏蔽敏感词。

3. 路由表和前缀匹配

• 网络路由表：在网络路由中，使用Trie（如Prefix Tree，前缀树）实现最长前缀匹配，快速确定数据包的转发路径。

4. 字符串统计与分析

• 统计字符串出现次数：在Trie的节点中维护计数器，可以统计字符串或前缀的出现频率，用于文本分析和数据挖掘。

5. DNA序列分析

• 生物信息学：DNA序列由A、C、G、T组成，利用Trie可以高效地存储和检索DNA序列片段，进行模式匹配和序列分析。

6. 压缩算法

• 压缩算法中的字典构建：如LZ前缀编码算法，利用Trie构建编码字典，实现数据的压缩。

7. 多模式串匹配

• Aho-Corasick算法：构建Trie树并结合失败指针，实现同时匹配多种模式串，应用于病毒检测、文本搜索等领域。

S2 示例

class TrieNode:
    """Trie 树的节点"""

    def __init__(self):
        self.children = {}  # 子节点字典，键为字符，值为 TrieNode
        self.is_end_of_word = False  # 标记是否为单词的结尾


class Trie:
    """Trie 树"""

    def __init__(self):
        self.root = TrieNode()

    def insert(self, word):
        """在 Trie 中插入一个单词"""
        node = self.root
        for char in word:
            if char not in node.children:
                # 如果子节点中没有当前字符，创建一个新的 TrieNode
                node.children[char] = TrieNode()
            node = node.children[char]
        # 单词结束，标记结尾
        node.is_end_of_word = True

    def search(self, word):
        """在 Trie 中搜索一个单词"""
        node = self.root
        for char in word:
            if char not in node.children:
                return False  # 未找到当前字符，返回 False
            node = node.children[char]
        return node.is_end_of_word  # 检查是否为单词的结尾

    def starts_with(self, prefix):
        """判断 Trie 中是否存在以指定前缀 prefix 开头的单词"""
        node = self.root
        for char in prefix:
            if char not in node.children:
                return False  # 未找到当前前缀
            node = node.children[char]
        return True  # 找到前缀

    def _traverse(self, node, prefix, words):
        """辅助函数，用于深度优先遍历 Trie，收集单词"""
        if node.is_end_of_word:
            words.append(prefix)
        for char, child_node in node.children.items():
            self._traverse(child_node, prefix + char, words)

    def get_words_with_prefix(self, prefix):
        """获取 Trie 中所有以指定前缀 prefix 开头的单词"""
        node = self.root
        for char in prefix:
            if char not in node.children:
                return []  # 前缀不存在，返回空列表
            node = node.children[char]
        words = []
        self._traverse(node, prefix, words)
        return words


# 测试代码
if __name__ == "__main__":
    trie = Trie()
    # 插入单词列表
    words_to_insert = ["apple", "app", "apply", "apt", "bat", "ball", "banana"]
    for word in words_to_insert:
        trie.insert(word)

    # 测试搜索单词
    words_to_search = ["app", "apple", "apt", "apart", "batman", "banana"]
    for word in words_to_search:
        found = trie.search(word)
        print(f"单词 '{word}' 在 Trie 中{'存在' if found else '不存在'}。")

    # 测试前缀查询
    prefixes = ["app", "ba", "cat"]
    for prefix in prefixes:
        has_prefix = trie.starts_with(prefix)
        print(f"Trie 中{'存在' if has_prefix else '不存在'}以 '{prefix}' 为前缀的单词。")
        if has_prefix:
            words_with_prefix = trie.get_words_with_prefix(prefix)
            print(f"以 '{prefix}' 为前缀的单词有：{words_with_prefix}")

结果

单词 'app' 在 Trie 中存在。
单词 'apple' 在 Trie 中存在。
单词 'apt' 在 Trie 中存在。
单词 'apart' 在 Trie 中不存在。
单词 'batman' 在 Trie 中不存在。
单词 'banana' 在 Trie 中存在。
Trie 中存在以 'app' 为前缀的单词。
以 'app' 为前缀的单词有：['app', 'apple', 'apply']
Trie 中存在以 'ba' 为前缀的单词。
以 'ba' 为前缀的单词有：['bat', 'ball', 'banana']
Trie 中不存在以 'cat' 为前缀的单词。

S3 应用1：基于 big.txt 实现单词的自动补全功能

需要下载包含大量英文单词和语料的 big.txt 文件，才能正常运行程序。该文件下载地址：big.txt

import re
from collections import defaultdict


class TrieNode:
    """Trie 树的节点"""

    def __init__(self):
        self.children = {}  # 子节点
        self.is_end_of_word = False  # 是否为完整单词
        self.frequency = 0  # 词频，用于排序
        self.word = None  # 完整单词


class AutoCompleteTrie:
    """自动补全功能的 Trie 树"""

    def __init__(self):
        self.root = TrieNode()

    def insert(self, word, frequency=1):
        """插入单词及其词频"""
        node = self.root
        for char in word:
            if char not in node.children:
                node.children[char] = TrieNode()
            node = node.children[char]
        node.is_end_of_word = True
        node.frequency += frequency
        node.word = word

    def search(self, prefix):
        """查找所有以 prefix 为前缀的单词"""
        node = self.root
        for char in prefix:
            if char not in node.children:
                return []  # 无匹配的前缀，返回空列表
            node = node.children[char]
        # 使用优先队列（堆）存储匹配的单词，按照词频排序
        words = []
        self._dfs(node, words)
        # 按照词频从高到低排序
        words.sort(key=lambda x: (-x[1], x[0]))
        return [word for word, freq in words]

    def _dfs(self, node, words):
        """深度优先搜索，收集单词及其词频"""
        if node.is_end_of_word:
            words.append((node.word, node.frequency))
        for child in node.children.values():
            self._dfs(child, words)


def preprocess_text(file_path):
    """读取文本文件，提取单词及其出现频率"""
    word_freq = defaultdict(int)
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            # 使用正则表达式提取单词，忽略大小写
            words = re.findall(r'\b[a-z]+\b', line.lower())
            for word in words:
                word_freq[word] += 1
    return word_freq


# 主程序
if __name__ == "__main__":
    trie = AutoCompleteTrie()
    # 从 big.txt 文件中提取单词及其频率
    word_freq_dict = preprocess_text('big.txt')
    print("正在构建 Trie 树，请稍候...")
    # 将单词及词频插入 Trie 树
    for word, freq in word_freq_dict.items():
        trie.insert(word, freq)
    print("Trie 树构建完成！")

    # 进入交互式查询
    while True:
        prefix = input("请输入搜索前缀（输入'exit'退出）：").strip().lower()
        if prefix == 'exit':
            break
        suggestions = trie.search(prefix)
        if suggestions:
            print("自动补全建议：")
            for word in suggestions[:10]:  # 只显示前 10 个建议
                print(f"- {word}")
        else:
            print("未找到匹配的建议。")

结果

正在构建 Trie 树，请稍候...
Trie 树构建完成！
请输入搜索前缀（输入'exit'退出）：ty
自动补全建议：
- type
- typical
- ty
- types
- typically
- tyranny
- tyrant
- tyre
请输入搜索前缀（输入'exit'退出）：an
自动补全建议：
- an
- anger
- answer
- analogy
- analysis
- ancestor
- ancient
- and
- angel
- angle
请输入搜索前缀（输入'exit'退出）：exit

S3 应用2：实现 IP 路由中的最长前缀匹配

class TrieNode:
    """Trie 树的节点，用于 IP 前缀匹配"""

    def __init__(self):
        self.children = {}
        self.next_hop = None  # 保存路由的下一跳信息


class IPRoutingTrie:
    """IP 路由的 Trie 实现"""

    def __init__(self):
        self.root = TrieNode()

    def insert(self, ip_prefix, next_hop):
        """
        插入路由前缀
        :param ip_prefix: 形如 '192.168.0.0/16'
        :param next_hop: 下一跳信息
        """
        ip, prefix_length = ip_prefix.split('/')
        prefix_length = int(prefix_length)
        binary_ip = self._ip_to_binary(ip)
        node = self.root
        for bit in binary_ip[:prefix_length]:
            if bit not in node.children:
                node.children[bit] = TrieNode()
            node = node.children[bit]
        node.next_hop = next_hop

    def search(self, ip_address):
        """查找目的 IP 地址的下一跳信息，使用最长前缀匹配"""
        binary_ip = self._ip_to_binary(ip_address)
        node = self.root
        last_match = None
        for bit in binary_ip:
            if bit in node.children:
                node = node.children[bit]
                if node.next_hop is not None:
                    last_match = node.next_hop
            else:
                break
        return last_match

    def _ip_to_binary(self, ip):
        """将 IP 地址转换为 32 位二进制字符串"""
        octets = ip.split('.')
        binary_ip = ''.join([format(int(octet), '08b') for octet in octets])
        return binary_ip


# 测试代码
if __name__ == "__main__":
    routing_table = IPRoutingTrie()
    # 添加路由前缀
    routing_entries = [
        ("192.168.0.0/16", "Router A"),
        ("192.168.1.0/24", "Router B"),
        ("10.0.0.0/8", "Router C"),
        ("0.0.0.0/0", "Default Gateway"),
    ]
    for prefix, next_hop in routing_entries:
        routing_table.insert(prefix, next_hop)

    # 查找目的 IP 地址的下一跳
    test_ips = [
        "192.168.1.100",
        "192.168.2.50",
        "10.1.2.3",
        "8.8.8.8",
    ]
    for ip in test_ips:
        next_hop = routing_table.search(ip)
        print(f"目的 IP {ip} 的下一跳为：{next_hop}")

结果

目的 IP 192.168.1.100 的下一跳为：Router B
目的 IP 192.168.2.50 的下一跳为：Router A
目的 IP 10.1.2.3 的下一跳为：Router C
目的 IP 8.8.8.8 的下一跳为：None

S3 应用3：基于 Trie 的压缩算法（LZW 算法）

class TrieNode:
    """Trie 节点，用于 LZW 压缩算法"""

    def __init__(self, index=None):
        self.children = {}  # 子节点
        self.index = index  # 节点对应的词典索引


def lzw_compress(uncompressed):
    """使用 LZW 算法进行压缩，使用 Trie 优化"""
    # 初始化词典，包含所有单字符
    dict_size = 256
    root = TrieNode()
    for i in range(dict_size):
        root.children[chr(i)] = TrieNode(index=i)

    result = []
    node = root
    w = ''
    for c in uncompressed:
        if c in node.children:
            node = node.children[c]
            w += c
        else:
            # 输出 w 的词典索引
            result.append(node.index)
            # 新的词典条目
            dict_size += 1
            node.children[c] = TrieNode(index=dict_size - 1)
            # 从根节点开始处理新字符
            node = root.children[c]
            w = c
    # 输出最后一个字符串的索引
    if w:
        result.append(node.index)
    return result


def lzw_decompress(compressed):
    """使用 LZW 算法进行解压"""
    # 初始化词典，包含所有单字符
    dict_size = 256
    dictionary = {i: chr(i) for i in range(dict_size)}

    result = []
    w = chr(compressed.pop(0))
    result.append(w)
    for k in compressed:
        if k in dictionary:
            entry = dictionary[k]
        elif k == dict_size:
            # 处理特殊情况
            entry = w + w[0]
        else:
            raise ValueError('Bad compressed k: %s' % k)
        result.append(entry)
        # 新的词典条目
        dictionary[dict_size] = w + entry[0]
        dict_size += 1
        w = entry
    return ''.join(result)


# 测试代码
if __name__ == "__main__":
    data = "TOBEORNOTTOBEORTOBEORNOT" * 3
    print("原始数据：", data)
    compressed = lzw_compress(data)
    print("压缩结果：", compressed)
    decompressed = lzw_decompress(compressed.copy())
    print("解压结果：", decompressed)
    print("压缩比：", len(compressed) / len(data))

结果

原始数据： TOBEORNOTTOBEORTOBEORNOTTOBEORNOTTOBEORTOBEORNOTTOBEORNOTTOBEORTOBEORNOT
压缩结果： [84, 79, 66, 69, 79, 82, 78, 79, 84, 256, 258, 260, 265, 259, 261, 263, 268, 260, 262, 264, 257, 269, 272, 270, 275, 266, 279, 278, 278, 274]
解压结果： TOBEORNOTTOBEORTOBEORNOTTOBEORNOTTOBEORTOBEORNOTTOBEORNOTTOBEORTOBEORNOT
压缩比： 0.4166666666666667