在 Python 3 中加速数百万个正则表达式替换

TLDR

如果您想要最快的基于正则表达式的解决方案，请使用此方法。对于类似于 OP 的数据集，它比接受的答案快大约 1000 倍。

如果您不关心正则表达式，请使用 this set-based version，它比正则表达式联合快 2000 倍。

使用 Trie 优化正则表达式

由于正则表达式引擎 doesn't do a very good job 优化了模式，simple Regex union 方法会因许多禁用词而变得缓慢。

可以使用所有被禁止的词创建一个 Trie 并编写相应的正则表达式。生成的 trie 或正则表达式并不是真正的人类可读的，但它们确实允许非常快速的查找和匹配。

例子

['foobar', 'foobah', 'fooxar', 'foozap', 'fooza']

https://i.stack.imgur.com/cFGZc.png

该列表被转换为一个 trie：

{
    'f': {
        'o': {
            'o': {
                'x': {
                    'a': {
                        'r': {
                            '': 1
                        }
                    }
                },
                'b': {
                    'a': {
                        'r': {
                            '': 1
                        },
                        'h': {
                            '': 1
                        }
                    }
                },
                'z': {
                    'a': {
                        '': 1,
                        'p': {
                            '': 1
                        }
                    }
                }
            }
        }
    }
}

然后到这个正则表达式模式：

r"\bfoo(?:ba[hr]|xar|zap?)\b"

https://i.stack.imgur.com/PoRCH.png

巨大的优势是要测试 zoo 是否匹配，正则表达式引擎只有 needs to compare the first character（它不匹配），而不是 trying the 5 words。这是 5 个单词的预处理过度杀伤力，但它显示了数千个单词的有希望的结果。

请注意，使用 (?:) non-capturing groups 是因为：

foobar|baz 将匹配 foobar 或 baz，但不匹配 foobaz

foo(bar|baz) 会将不需要的信息保存到捕获组。

代码

这是一个稍作修改的 gist，我们可以将其用作 trie.py 库：

import re


class Trie():
    """Regex::Trie in Python. Creates a Trie out of a list of words. The trie can be exported to a Regex pattern.
    The corresponding Regex should match much faster than a simple Regex union."""

    def __init__(self):
        self.data = {}

    def add(self, word):
        ref = self.data
        for char in word:
            ref[char] = char in ref and ref[char] or {}
            ref = ref[char]
        ref[''] = 1

    def dump(self):
        return self.data

    def quote(self, char):
        return re.escape(char)

    def _pattern(self, pData):
        data = pData
        if "" in data and len(data.keys()) == 1:
            return None

        alt = []
        cc = []
        q = 0
        for char in sorted(data.keys()):
            if isinstance(data[char], dict):
                try:
                    recurse = self._pattern(data[char])
                    alt.append(self.quote(char) + recurse)
                except:
                    cc.append(self.quote(char))
            else:
                q = 1
        cconly = not len(alt) > 0

        if len(cc) > 0:
            if len(cc) == 1:
                alt.append(cc[0])
            else:
                alt.append('[' + ''.join(cc) + ']')

        if len(alt) == 1:
            result = alt[0]
        else:
            result = "(?:" + "|".join(alt) + ")"

        if q:
            if cconly:
                result += "?"
            else:
                result = "(?:%s)?" % result
        return result

    def pattern(self):
        return self._pattern(self.dump())

测试

这是一个小测试（与 this one 相同）：

# Encoding: utf-8
import re
import timeit
import random
from trie import Trie

with open('/usr/share/dict/american-english') as wordbook:
    banned_words = [word.strip().lower() for word in wordbook]
    random.shuffle(banned_words)

test_words = [
    ("Surely not a word", "#surely_NöTäWORD_so_regex_engine_can_return_fast"),
    ("First word", banned_words[0]),
    ("Last word", banned_words[-1]),
    ("Almost a word", "couldbeaword")
]

def trie_regex_from_words(words):
    trie = Trie()
    for word in words:
        trie.add(word)
    return re.compile(r"\b" + trie.pattern() + r"\b", re.IGNORECASE)

def find(word):
    def fun():
        return union.match(word)
    return fun

for exp in range(1, 6):
    print("\nTrieRegex of %d words" % 10**exp)
    union = trie_regex_from_words(banned_words[:10**exp])
    for description, test_word in test_words:
        time = timeit.timeit(find(test_word), number=1000) * 1000
        print("  %s : %.1fms" % (description, time))

它输出：

TrieRegex of 10 words
  Surely not a word : 0.3ms
  First word : 0.4ms
  Last word : 0.5ms
  Almost a word : 0.5ms

TrieRegex of 100 words
  Surely not a word : 0.3ms
  First word : 0.5ms
  Last word : 0.9ms
  Almost a word : 0.6ms

TrieRegex of 1000 words
  Surely not a word : 0.3ms
  First word : 0.7ms
  Last word : 0.9ms
  Almost a word : 1.1ms

TrieRegex of 10000 words
  Surely not a word : 0.1ms
  First word : 1.0ms
  Last word : 1.2ms
  Almost a word : 1.2ms

TrieRegex of 100000 words
  Surely not a word : 0.3ms
  First word : 1.2ms
  Last word : 0.9ms
  Almost a word : 1.6ms

有关信息，正则表达式的开头如下：

(?:a(?:(?:\'s|a(?:\'s|chen|liyah(?:\'s)?|r(?:dvark(?:(?:\'s|s)) ))?|on))|b(?:\'s|a(?:c(?:us(?:(?:\'s|es)))?|[ik])|ft|lone(? :(?:\'s|s))?|ndon(?:(?:ed|ing|ment(?:\'s)?|s))?|s(?:e(?:(?: ment(?:\'s)?|[ds]))?|h(?:(?:e[ds]|ing))?|ing)|t(?:e(?:(?:ment( ?:\'s)?|[ds]))?|ing|toir(?:(?:\'s|s))?))|b(?:as(?:id)?|e(? :ss(?:(?:\'s|es))?|y(?:(?:\'s|s))?)|ot(?:(?:\'s|t(?:\ 's)?|s))?|reviat(?:e[ds]?|i(?:ng|on(?:(?:\'s|s))?))|y(?:\' s)?|\é(?:(?:\'s|s))?)|d(?:icat(?:e[ds]?|i(?:ng|on(?:(?:\ 's|s))?))|om(?:en(?:(?:\'s|s))?|inal)|u(?:ct(?:(?:ed|i(?: ng|on(?:(?:\'s|s))?)|or(?:(?:\'s|s))?|s))?|l(?:\'s)?) )|e(?:(?:\'s|am|l(?:(?:\'s|ard|son(?:\'s)?))?|r(?:deen(?:\ 's)?|nathy(?:\'s)?|ra(?:nt|tion(?:(?:\'s|s))?))|t(?:(?:t(?: e(?:r(?:(?:\'s|s))?|d)|ing|or(?:(?:\'s|s))?)|s))?|yance(? :\'s)?|d))?|hor(?:(?:r(?:e(?:n(?:ce(?:\'s)?|t)|d)|ing)| s))?|i(?:d(?:e[ds]?|ing|jan(?:\'s)?)|gail|l(?:ene|it(?:ies|y(?: \'s)?)))|j(?:ect(?:ly)?|ur(?:ation(?:(?:\'s|s))?|e[ds]?|ing)) |l(?:a(?:tive(?:(?:\'s|s))?|ze)|e(?:(?:st|r))?|oom|ution(?:(? :\'s|s))?|y )|m\'s|n(?:e(?:gat(?:e[ds]?|i(?:ng|on(?:\'s)?))|r(?:\'s )?)|正规(?:(?:it(?:ies|y(?:\'s)?)|ly))?)|o(?:ard|de(?:(?:\'s |s))?|li(?:sh(?:(?:e[ds]|ing))?|tion(?:(?:\'s|ist(?:(?:\'s|s ))?))?)|mina(?:bl[ey]|t(?:e[ds]?|i(?:ng|on(?:(?:\'s|s))?)) )|r(?:igin(?:al(?:(?:\'s|s))?|e(?:(?:\'s|s))?)|t(?:(?: ed|i(?:ng|on(?:(?:\'s|ist(?:(?:\'s|s))?|s))?|ve)|s))?)|u (?:nd(?:(?:ed|ing|s))?|t)|ve(?:(?:\'s|board))?)|r(?:a(?:cadabra(? :\'s)?|d(?:e[ds]?|ing)|ham(?:\'s)?|m(?:(?:\'s|s))?|si(?: on(?:(?:\'s|s))?|ve(?:(?:\'s|ly|ness(?:\'s)?|s))?))|east|idg( ?:e(?:(?:ment(?:(?:\'s|s))?|[ds]))?|ing|ment(?:(?:\'s|s))?) |o(?:ad|gat(?:e[ds]?|i(?:ng|on(?:(?:\'s|s))?)))|upt(?:(?:e (?:st|r)|ly|ness(?:\'s)?))?)|s(?:alom|c(?:ess(?:(?:\'s|e[ds]| ing))?|issa(?:(?:\'s|[es]))?|ond(?:(?:ed|ing|s))?)|en(?:ce(?:(? :\'s|s))?|t(?:(?:e(?:e(?:(?:\'s|ism(?:\'s)?|s))?|d)| ing|ly|s))?)|inth(?:(?:\'s|e(?:\'s)?))?|o(?:l(?:ut(?:e(?: (?:\'s|ly|st?))?|i(?:on(?:\'s)?|sm(?:\'s)?))|v(?:e[ds]? |ing))|r(?:b(?:(?:e(?:n(?:cy(?:\'s)?|t(?:(?:\'s|s)))?) |d)|ing|s))?|pti ...

真的很难读，但是对于 100000 个禁用词的列表，这个 Trie 正则表达式比简单的正则表达式联合快 1000 倍！

这是完整的 trie 图表，使用 trie-python-graphviz 和 graphviz twopi 导出：

https://i.stack.imgur.com/gpFLp.jpg

---

TLDR

如果您想要最快的解决方案，请使用此方法（使用集合查找）。对于类似于 OP 的数据集，它比接受的答案快大约 2000 倍。

如果您坚持使用正则表达式进行查找，请使用 this trie-based version，它仍然比正则表达式联合快 1000 倍。

理论

如果您的句子不是巨大的字符串，那么每秒处理超过 50 个可能是可行的。

如果您将所有被禁止的单词保存到一个集合中，则可以非常快速地检查该集合中是否包含另一个单词。

将逻辑打包到一个函数中，将此函数作为 re.sub 的参数，你就完成了！

代码

import re
with open('/usr/share/dict/american-english') as wordbook:
    banned_words = set(word.strip().lower() for word in wordbook)


def delete_banned_words(matchobj):
    word = matchobj.group(0)
    if word.lower() in banned_words:
        return ""
    else:
        return word

sentences = ["I'm eric. Welcome here!", "Another boring sentence.",
             "GiraffeElephantBoat", "sfgsdg sdwerha aswertwe"] * 250000

word_pattern = re.compile('\w+')

for sentence in sentences:
    sentence = word_pattern.sub(delete_banned_words, sentence)

转换后的句子是：

' .  !
  .
GiraffeElephantBoat
sfgsdg sdwerha aswertwe

注意：

搜索不区分大小写（感谢 lower()）

用“”替换单词可能会留下两个空格（如您的代码中所示）

对于 python3，\w+ 也匹配重音字符（例如“ångström”）。

任何非单词字符（制表符、空格、换行符、标记……）都将保持不变。

表现

有一百万个句子，banned_words 有近 100000 个单词，脚本运行时间不到 7 秒。

相比之下，Liteye 的 answer 需要 160 秒才能写 10000 句。

n 是词的总数量，m 是禁止词的数量，OP 和 Liteye 的代码是 O(n*m)。

相比之下，我的代码应该在 O(n+m) 中运行。考虑到句子多于禁词，算法变为O(n)。

正则表达式联合测试

具有 '\b(word1|word2|...|wordN)\b' 模式的正则表达式搜索的复杂性是多少？是 O(N) 还是 O(1)？

很难掌握正则表达式引擎的工作方式，所以让我们编写一个简单的测试。

此代码将 10**i 个随机英文单词提取到一个列表中。它创建相应的正则表达式联合，并用不同的词对其进行测试：

one 显然不是一个词（它以 # 开头）

一个是列表中的第一个单词

一个是列表中的最后一个单词

一个看起来像一个词，但不是

import re
import timeit
import random

with open('/usr/share/dict/american-english') as wordbook:
    english_words = [word.strip().lower() for word in wordbook]
    random.shuffle(english_words)

print("First 10 words :")
print(english_words[:10])

test_words = [
    ("Surely not a word", "#surely_NöTäWORD_so_regex_engine_can_return_fast"),
    ("First word", english_words[0]),
    ("Last word", english_words[-1]),
    ("Almost a word", "couldbeaword")
]


def find(word):
    def fun():
        return union.match(word)
    return fun

for exp in range(1, 6):
    print("\nUnion of %d words" % 10**exp)
    union = re.compile(r"\b(%s)\b" % '|'.join(english_words[:10**exp]))
    for description, test_word in test_words:
        time = timeit.timeit(find(test_word), number=1000) * 1000
        print("  %-17s : %.1fms" % (description, time))

它输出：

First 10 words :
["geritol's", "sunstroke's", 'fib', 'fergus', 'charms', 'canning', 'supervisor', 'fallaciously', "heritage's", 'pastime']

Union of 10 words
  Surely not a word : 0.7ms
  First word        : 0.8ms
  Last word         : 0.7ms
  Almost a word     : 0.7ms

Union of 100 words
  Surely not a word : 0.7ms
  First word        : 1.1ms
  Last word         : 1.2ms
  Almost a word     : 1.2ms

Union of 1000 words
  Surely not a word : 0.7ms
  First word        : 0.8ms
  Last word         : 9.6ms
  Almost a word     : 10.1ms

Union of 10000 words
  Surely not a word : 1.4ms
  First word        : 1.8ms
  Last word         : 96.3ms
  Almost a word     : 116.6ms

Union of 100000 words
  Surely not a word : 0.7ms
  First word        : 0.8ms
  Last word         : 1227.1ms
  Almost a word     : 1404.1ms

所以看起来搜索具有 '\b(word1|word2|...|wordN)\b' 模式的单个单词具有：

O(1) 最佳情况

O(n/2) 平均情况，仍然是 O(n)

O(n) 最坏情况

这些结果与简单的循环搜索一致。

一个比正则表达式联合更快的替代方法是创建 regex pattern from a trie。

---

您可以尝试的一件事是编译一个单一模式，例如 "\b(word1|word2|word3)\b"。

因为 re 依赖 C 代码来进行实际匹配，所以节省的费用会非常显着。

正如@pvg 在评论中指出的那样，它也受益于单通道匹配。

如果您的话不是正则表达式，Eric 的 answer 会更快。

---

您可能想尝试的一件事是预处理句子以对单词边界进行编码。基本上通过在单词边界上拆分将每个句子变成单词列表。

这应该更快，因为要处理一个句子，您只需要逐步检查每个单词并检查它是否匹配。

目前，正则表达式搜索每次都必须再次遍历整个字符串，寻找单词边界，然后在下一次遍历之前“丢弃”这项工作的结果。

---

好吧，这是一个快速简便的解决方案，带有测试集。

最佳策略：

re.sub("\w+",repl,sentence) 搜索单词。

“repl”可以是可调用的。我使用了一个执行字典查找的函数，字典包含要搜索和替换的单词。

这是最简单和最快的解决方案（参见下面示例代码中的函数 replace4）。

次优策略：

这个想法是使用 re.split 将句子拆分为单词，同时保留分隔符以便稍后重建句子。然后，通过简单的 dict 查找来完成替换。

实现：（参见下面示例代码中的函数 replace3）。

示例功能的时序：

replace1:     0.62 sentences/s
replace2:     7.43 sentences/s
replace3: 48498.03 sentences/s
replace4: 61374.97 sentences/s (...and 240,000/s with PyPy)

...和代码：

#! /bin/env python3
# -*- coding: utf-8

import time, random, re

def replace1( sentences ):
    for n, sentence in enumerate( sentences ):
        for search, repl in patterns:
            sentence = re.sub( "\\b"+search+"\\b", repl, sentence )

def replace2( sentences ):
    for n, sentence in enumerate( sentences ):
        for search, repl in patterns_comp:
            sentence = re.sub( search, repl, sentence )

def replace3( sentences ):
    pd = patterns_dict.get
    for n, sentence in enumerate( sentences ):
        #~ print( n, sentence )
        # Split the sentence on non-word characters.
        # Note: () in split patterns ensure the non-word characters ARE kept
        # and returned in the result list, so we don't mangle the sentence.
        # If ALL separators are spaces, use string.split instead or something.
        # Example:
        #~ >>> re.split(r"([^\w]+)", "ab céé? . d2eéf")
        #~ ['ab', ' ', 'céé', '? . ', 'd2eéf']
        words = re.split(r"([^\w]+)", sentence)
        
        # and... done.
        sentence = "".join( pd(w,w) for w in words )

        #~ print( n, sentence )

def replace4( sentences ):
    pd = patterns_dict.get
    def repl(m):
        w = m.group()
        return pd(w,w)
    
    for n, sentence in enumerate( sentences ):
        sentence = re.sub(r"\w+", repl, sentence)



# Build test set
test_words = [ ("word%d" % _) for _ in range(50000) ]
test_sentences = [ " ".join( random.sample( test_words, 10 )) for _ in range(1000) ]

# Create search and replace patterns
patterns = [ (("word%d" % _), ("repl%d" % _)) for _ in range(20000) ]
patterns_dict = dict( patterns )
patterns_comp = [ (re.compile("\\b"+search+"\\b"), repl) for search, repl in patterns ]


def test( func, num ):
    t = time.time()
    func( test_sentences[:num] )
    print( "%30s: %.02f sentences/s" % (func.__name__, num/(time.time()-t)))

print( "Sentences", len(test_sentences) )
print( "Words    ", len(test_words) )

test( replace1, 1 )
test( replace2, 10 )
test( replace3, 1000 )
test( replace4, 1000 )

编辑：您也可以在检查是否传递小写的句子列表并编辑 repl 时忽略小写

def replace4( sentences ):
pd = patterns_dict.get
def repl(m):
    w = m.group()
    return pd(w.lower(),w)

---

也许 Python 在这里不是正确的工具。这是一个带有 Unix 工具链的

sed G file         |
tr ' ' '\n'        |
grep -vf blacklist |
awk -v RS= -v OFS=' ' '{$1=$1}1'

假设您的黑名单文件已添加单词边界进行预处理。步骤是：将文件转换为双倍行距，将每个句子拆分为每行一个单词，从文件中批量删除黑名单单词，然后合并行。

这应该至少快一个数量级。

用于从单词中预处理黑名单文件（每行一个单词）

sed 's/.*/\\b&\\b/' words > blacklist

---

这个怎么样：

#!/usr/bin/env python3

from __future__ import unicode_literals, print_function
import re
import time
import io

def replace_sentences_1(sentences, banned_words):
    # faster on CPython, but does not use \b as the word separator
    # so result is slightly different than replace_sentences_2()
    def filter_sentence(sentence):
        words = WORD_SPLITTER.split(sentence)
        words_iter = iter(words)
        for word in words_iter:
            norm_word = word.lower()
            if norm_word not in banned_words:
                yield word
            yield next(words_iter) # yield the word separator

    WORD_SPLITTER = re.compile(r'(\W+)')
    banned_words = set(banned_words)
    for sentence in sentences:
        yield ''.join(filter_sentence(sentence))


def replace_sentences_2(sentences, banned_words):
    # slower on CPython, uses \b as separator
    def filter_sentence(sentence):
        boundaries = WORD_BOUNDARY.finditer(sentence)
        current_boundary = 0
        while True:
            last_word_boundary, current_boundary = current_boundary, next(boundaries).start()
            yield sentence[last_word_boundary:current_boundary] # yield the separators
            last_word_boundary, current_boundary = current_boundary, next(boundaries).start()
            word = sentence[last_word_boundary:current_boundary]
            norm_word = word.lower()
            if norm_word not in banned_words:
                yield word

    WORD_BOUNDARY = re.compile(r'\b')
    banned_words = set(banned_words)
    for sentence in sentences:
        yield ''.join(filter_sentence(sentence))


corpus = io.open('corpus2.txt').read()
banned_words = [l.lower() for l in open('banned_words.txt').read().splitlines()]
sentences = corpus.split('. ')
output = io.open('output.txt', 'wb')
print('number of sentences:', len(sentences))
start = time.time()
for sentence in replace_sentences_1(sentences, banned_words):
    output.write(sentence.encode('utf-8'))
    output.write(b' .')
print('time:', time.time() - start)

这些解决方案在单词边界上拆分并查找集合中的每个单词。它们应该比单词替换的 re.sub 更快（Liteyes 的解决方案），因为这些解决方案是 O(n)，其中 n 是由于 amortized O(1) 集查找而导致的输入大小，而使用正则表达式替换会导致正则表达式引擎必须检查每个字符上的单词匹配，而不仅仅是单词边界。我的解决方案特别注意保留原始文本中使用的空格（即它不压缩空格并保留制表符、换行符和其他空格字符），但如果您决定不关心它，它从输出中删除它们应该相当简单。

我在 corpus.txt 上进行了测试，它是从 Gutenberg 项目下载的多本电子书的串联，banned_words.txt 是从 Ubuntu 的单词列表（/usr/share/dict/american-english）中随机挑选的 20000 个单词。处理 862462 个句子大约需要 30 秒（在 PyPy 上是一半）。我将句子定义为以“。”分隔的任何内容。

$ # replace_sentences_1()
$ python3 filter_words.py 
number of sentences: 862462
time: 24.46173644065857
$ pypy filter_words.py 
number of sentences: 862462
time: 15.9370770454

$ # replace_sentences_2()
$ python3 filter_words.py 
number of sentences: 862462
time: 40.2742919921875
$ pypy filter_words.py 
number of sentences: 862462
time: 13.1190629005

PyPy 尤其从第二种方法中受益更多，而 CPython 在第一种方法中表现更好。上面的代码应该适用于 Python 2 和 3。

---

实用方法

下面描述的解决方案使用大量内存来将所有文本存储在同一字符串中并降低复杂度。如果 RAM 是一个问题，请在使用之前三思而后行。

使用 join/split 技巧，您可以完全避免循环，从而加快算法速度。

用不包含在句子中的特殊分隔符连接句子：

merged_sentences = ' * '.join(sentences)

使用 | 为您需要从句子中删除的所有单词编译一个正则表达式“或”正则表达式：

regex = re.compile(r'\b({})\b'.format('|'.join(words)), re.I) # re.I is a case insensitive flag

用编译的正则表达式为单词下标，并用特殊分隔符将其拆分回分隔的句子：

clean_sentences = re.sub(regex, "", merged_sentences).split(' * ')

表现

"".join 复杂度为 O(n)。这是非常直观的，但无论如何都有一个来源的简短引用：

for (i = 0; i < seqlen; i++) {
    [...]
    sz += PyUnicode_GET_LENGTH(item);

因此，对于 join/split，您有 O(words) + 2*O(sentences) 与初始方法的 2*O(N²) 相比，它仍然是线性复杂度。

顺便说一句，不要使用多线程。 GIL 将阻止每个操作，因为您的任务严格受 CPU 限制，因此 GIL 没有机会被释放，但每个线程将同时发送滴答声，这会导致额外的工作，甚至导致操作无穷大。

---

将所有句子连接到一个文档中。使用 Aho-Corasick 算法 (here's one) 的任何实现来定位所有“坏”词。遍历文件，替换每个坏词，更新后面找到的词的偏移量等。

---