Python中快速解析文本词组组合的高效优化方案

原代码的核心问题是生成所有可能的单词组合，时间复杂度为O(2^n)（n为文本单词数），当文本单词数达到1000时，组合数会呈天文数字级增长，处理百万行数据自然会极慢。以下是两种高效替代方案：

方案一：分组滑动窗口（简单高效，适合新手） #

思路 #

1. 预处理字典：将短语按「单词长度」分组，同时建立小写单词元组→原短语的映射，解决大小写不敏感和标点问题。
2. 遍历文本时，仅检查字典中存在的短语长度对应的连续单词窗口，避免生成无用的非连续组合。

代码实现 #

from string import punctuation

def preprocess_dict(phrase_list):
    # 按短语的单词长度分组，存储(小写单词元组: 原短语)的映射
    length_to_phrases = {}
    for phrase in phrase_list:
        # 处理短语：拆分、去标点、转小写
        words = [word.strip(punctuation).lower() for word in phrase.split()]
        phrase_len = len(words)
        if phrase_len not in length_to_phrases:
            length_to_phrases[phrase_len] = {}
        # 用元组作为键（列表不可哈希）
        length_to_phrases[phrase_len][tuple(words)] = phrase
    return length_to_phrases

def retrieve_text_from_body(body, processed_dict):
    result = []
    # 处理文本：拆分、去标点、转小写
    body_words = [word.strip(punctuation).lower() for word in body.split()]
    body_len = len(body_words)
    
    # 遍历字典中存在的所有短语长度
    for phrase_len, phrase_map in processed_dict.items():
        if phrase_len > body_len:
            continue  # 文本长度不足，跳过
        # 滑动窗口遍历所有连续phrase_len个单词的组合
        for i in range(body_len - phrase_len + 1):
            current_tuple = tuple(body_words[i:i+phrase_len])
            if current_tuple in phrase_map:
                result.append(phrase_map[current_tuple])
    # 去重（避免同一短语被多次匹配）
    return list(set(result))

# 示例使用
if __name__ == "__main__":
    body = "The big fat cat sits"
    dict_list = ["Big Fat", "Little Mouse", "Cat Sits"]
    processed_dict = preprocess_dict(dict_list)
    print(retrieve_text_from_body(body, processed_dict))  # 输出: ['Big Fat', 'Cat Sits']

效率提升原因 #

• 时间复杂度降至O(m*k)：m为文本单词数，k为字典中短语的最大长度，远低于原代码的指数级复杂度。
• 仅生成有意义的连续单词组合，避免了原代码中大量无用的非连续组合（比如['the','fat']这类不可能匹配的组合）。

方案二：Aho-Corasick自动机（超大规模字典/文本首选） #

如果字典短语数量极多（十万级以上），滑动窗口的效率可能仍不足，此时可以用Aho-Corasick多模式匹配算法，它能在O(n + m + z)的时间内完成匹配（n为文本长度，m为所有短语总长度，z为匹配次数），适合批量处理百万行数据。

代码实现 #

需要先安装依赖库：pip install pyahocorasick

import ahocorasick
from string import punctuation

def build_automaton(phrase_list):
    automaton = ahocorasick.Automaton()
    # 预处理短语并加入自动机
    for idx, phrase in enumerate(phrase_list):
        # 转换为小写、去标点的空格分隔字符串
        processed_phrase = ' '.join([word.strip(punctuation).lower() for word in phrase.split()])
        automaton.add_word(processed_phrase, (idx, phrase))
    automaton.make_automaton()
    return automaton

def retrieve_text_from_body(body, automaton):
    result = set()
    # 处理文本：转换为小写、去标点的空格分隔字符串
    processed_body = ' '.join([word.strip(punctuation).lower() for word in body.split()])
    # 遍历所有匹配结果
    for end_idx, (idx, original_phrase) in automaton.iter(processed_body):
        result.add(original_phrase)
    return list(result)

# 示例使用
if __name__ == "__main__":
    body = "The big fat cat sits"
    dict_list = ["Big Fat", "Little Mouse", "Cat Sits"]
    automaton = build_automaton(dict_list)
    print(retrieve_text_from_body(body, automaton))  # 输出: ['Big Fat', 'Cat Sits']

优势 #

• 一次遍历文本即可找到所有匹配短语，适合字典短语数量大、文本批量处理的场景。
• 避免了滑动窗口中对不同长度短语的多次遍历。

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