Фильтр чата: автомат Ахо-Корасик за один проход

Если ты умеешь построить автомат Ахо-Корасик и знаешь, зачем считать слова по цепочке ссылок неудач, листай до «Теперь сам».

Как думаешь, сколько стоит фильтр

Модератор ведёт список из 2000 запрещённых слов, в чат приходит 50 сообщений в секунду. Наивный фильтр ищет каждое слово в каждом сообщении отдельно. Сколько раз он прочитает одно сообщение? А сколько должен?

Ответь себе до чтения. Наивный фильтр читает сообщение 2000 раз - по разу на слово. При этом все слова ищутся в одном и том же тексте, и большинство из них начинаются одинаково. Достаточно читать сообщение ОДИН раз, держа в голове сразу все слова, которые совпадают с прочитанным хвостом текста. Это и делает автомат: его состояние - «какой самый длинный кусок какого-нибудь слова сейчас совпал».

На что это похоже

Опытный корректор, который вычитывает текст на сотню запрещённых оборотов. Он не читает текст сто раз - он читает один раз и замечает любой оборот, потому что держит в голове их все и помнит, на каком слове каждого сейчас остановился. Если начало одного оборота сорвалось, он не возвращается назад, а сразу переключается на другой оборот, который продолжает уже прочитанное.

Механизм: бор, ссылки неудач, выходы

Автомат строится из трёх частей.

Бор - дерево префиксов всех слов: из корня по буквам, каждая вершина - префикс какого-то слова. В вершине, где кончается слово, ставится счётчик. Ссылка неудачи вершины ведёт в вершину самого длинного собственного суффикса её префикса, который тоже есть в боре: если следующая буква текста не продолжает текущий префикс, автомат переходит по ссылке и пробует продолжить там. Ссылки строятся обходом бора в ширину: ссылка вершины считается из ссылки её родителя.

Третья часть - выходы. Когда автомат стоит в вершине «ворон», в этой же позиции текста кончаются и все слова, которые являются суффиксами «ворон» и есть в списке: «рон», «он». Они лежат на цепочке ссылок неудач от вершины. Поэтому каждой вершине сразу при построении считают сколько слов кончается здесь с учётом цепочки: vsego[u] = konec[u] + vsego[неудача(u)] - ссылка ведёт в вершину меньшей глубины, а обход в ширину обрабатывает её раньше.

Чтение текста: на каждую букву автомат либо идёт по переходу бора, либо по ссылкам неудач до вершины, где переход есть, - или остаётся в корне, если этой буквы нет ни в одном слове, как у пробела. Переходов по ссылкам за всё время не больше, чем переходов вперёд, - каждый уменьшает глубину, а вперёд глубина растёт на 1 за букву. Итог - линия от длины текста плюс число найденных вхождений.

Правило

Много запрещённых слов ищут автоматом Ахо-Корасик: бор, ссылки неудач, счётчик слов с учётом цепочки ссылок. Автомат строится один раз на список, а каждое сообщение читается один раз, сколько бы слов ни было. Без суммы по цепочке пропадают слова, которые кончаются внутри других.

Разбор: сорок слов, один проход

Чат из 60 «слов», 40 запрещённых двухслоговых сочетаний. Наивный поиск против автомата - считаем сравнения и шаги.

import random

random.seed(23)
slogi = ["ka", "ro", "mi", "lu", "ten", "vor", "pa", "zu", "dri", "os"]

def slovo(n):
    w = ""
    for i in range(n):
        w = w + slogi[random.randint(0, 9)]
    return w

zapret = []
while len(zapret) < 40:
    w = slovo(2)
    if w not in zapret:
        zapret.append(w)
chat = ""
for i in range(60):
    chat = chat + slovo(random.randint(1, 3)) + " "
rabota = [0, 0]

def naivno():
    # каждое запрещённое слово ищем по всему тексту отдельно
    najdeno = 0
    for w in zapret:
        for i in range(len(chat) - len(w) + 1):
            j = 0
            while j < len(w):
                rabota[0] = rabota[0] + 1
                if chat[i + j] != w[j]:
                    break
                j = j + 1
            if j == len(w):
                najdeno = najdeno + 1
    return najdeno

# автомат Ахо-Корасик (раздел 17): бор слов, ссылки неудач и выходы
perehod = [{}]
konec = [0]                 # сколько слов кончается в этой вершине
for w in zapret:
    v = 0
    for ch in w:
        if ch not in perehod[v]:
            perehod.append({})
            konec.append(0)
            perehod[v][ch] = len(perehod) - 1
        v = perehod[v][ch]
    konec[v] = konec[v] + 1
neudacha = [0] * len(perehod)
vsego = [0] * len(perehod)  # слов, кончающихся здесь, по всей цепочке
och = []
for ch in perehod[0]:
    och.append(perehod[0][ch])
for u in och:
    vsego[u] = konec[u]
golova = 0
while golova < len(och):
    v = och[golova]
    golova = golova + 1
    for ch in perehod[v]:
        u = perehod[v][ch]
        f = neudacha[v]
        while f > 0 and ch not in perehod[f]:
            f = neudacha[f]
        if ch in perehod[f] and perehod[f][ch] != u:
            neudacha[u] = perehod[f][ch]
        vsego[u] = konec[u] + vsego[neudacha[u]]
        och.append(u)

def avtomatom():
    v = 0
    najdeno = 0
    for ch in chat:
        rabota[1] = rabota[1] + 1
        while v > 0 and ch not in perehod[v]:
            rabota[1] = rabota[1] + 1
            v = neudacha[v]
        if ch in perehod[v]:
            v = perehod[v][ch]
        najdeno = najdeno + vsego[v]
    return najdeno

print("текст:", len(chat), "символов, запрещённых слов:", len(zapret))
print("наивно:  ", naivno(), "вхождений |", rabota[0], "сравнений")
print("автоматом:", avtomatom(), "вхождений |", rabota[1], "шагов")

текст: 348 символов, запрещённых слов: 40
наивно:   26 вхождений | 15178 сравнений
автоматом: 26 вхождений | 489 шагов

Оба нашли 26 вхождений. Наивный поиск сделал 15 178 сравнений - по проходу текста на каждое слово; автомат - 489 шагов: 348 переходов по буквам и 141 переход по ссылкам неудач. При 2000 словах наивный поиск вырастет в 50 раз, а для автомата число слов не поднимает границу: переходов по буквам столько же, а переходов по ссылкам по-прежнему не больше длины текста. Граница правила: стоимость переехала в построение. Бор и ссылки строятся за сумму длин всех слов, и при каждом изменении списка - заново. Модератор меняет список раз в день, сообщения идут пятьдесят в секунду, - обмен выгодный; если бы список менялся на каждое сообщение, он был бы бессмысленным.

Разбор: «он» внутри «ворона»

Три слова, которые кончаются друг в друге: «ворон», «рон», «он». Текст «вороненок и ворон» содержит каждое дважды - всего шесть вхождений. Считаем двумя способами: только слово самой вершины и сумму по цепочке ссылок неудач.

def postroit(slova):
    # бор, ссылки неудач и vsego - сколько слов кончается в вершине
    # с учётом её суффиксов по цепочке неудач
    perehod = [{}]
    konec = [0]
    for w in slova:
        v = 0
        for ch in w:
            if ch not in perehod[v]:
                perehod.append({})
                konec.append(0)
                perehod[v][ch] = len(perehod) - 1
            v = perehod[v][ch]
        konec[v] = konec[v] + 1
    neudacha = [0] * len(perehod)
    vsego = [0] * len(perehod)
    och = []
    for ch in perehod[0]:
        och.append(perehod[0][ch])
    for u in och:
        vsego[u] = konec[u]
    golova = 0
    while golova < len(och):
        v = och[golova]
        golova = golova + 1
        for ch in perehod[v]:
            u = perehod[v][ch]
            f = neudacha[v]
            while f > 0 and ch not in perehod[f]:
                f = neudacha[f]
            if ch in perehod[f] and perehod[f][ch] != u:
                neudacha[u] = perehod[f][ch]
            vsego[u] = konec[u] + vsego[neudacha[u]]
            och.append(u)
    return [perehod, neudacha, konec, vsego]

def schitat(avtomat, tekst, po_cepochke):
    perehod = avtomat[0]
    neudacha = avtomat[1]
    v = 0
    najdeno = 0
    for ch in tekst:
        while v > 0 and ch not in perehod[v]:
            v = neudacha[v]
        if ch in perehod[v]:
            v = perehod[v][ch]
        if po_cepochke:
            najdeno = najdeno + avtomat[3][v]    # vsego: с суффиксами
        else:
            najdeno = najdeno + avtomat[2][v]    # konec: своё слово
    return najdeno

a = postroit(["ворон", "рон", "он"])
tekst = "вороненок и ворон"
print("только слово вершины:", schitat(a, tekst, False))
print("по цепочке неудач:   ", schitat(a, tekst, True))

только слово вершины: 2
по цепочке неудач:    6

Автомат, который смотрит только на слово своей вершины, нашёл два «ворона» и пропустил «рон» и «он»: в позиции после «н» он стоял в вершине «ворон», а не в «он», и счётчик «он» в этот момент не смотрел. Сумма по цепочке собирает все слова, которые кончаются в этой позиции, - шесть. Это классическая ошибка самописных фильтров: в тестах слова не вкладываются друг в друга, а в живом чате - постоянно. Граница правила: если нужно не число вхождений, а сами слова (чтобы закрыть их звёздочками), по цепочке идут только через вершины, где слова действительно кончаются, - для этого хранят ссылку «ближайший суффикс, который сам является словом», иначе цепочка ссылок может быть длинной и пустой.

Что ломается без этого

Фильтр, который тормозит чат. Список запретов вырос до тысяч слов, наивный фильтр читает каждое сообщение тысячи раз - и чат отстаёт на секунды.

Пропущенное вложенное слово. Фильтр закрыл «ворона», а «он» внутри прошёл - потому что выход по цепочке не учтён.

Автомат на каждое сообщение. Автомат пересобирают при каждом сообщении «на всякий случай» - и экономия съедена построением.

Зачем это в работе

Антивирусы сверяют файлы с тысячами сигнатур, системы обнаружения вторжений ищут образцы атак в сетевом трафике, grep -F с файлом образцов ищет тысячи строк за проход - всё это автоматы Ахо-Корасик или их родственники. Модель везде одна: дорогое построение по списку один раз, дешёвое чтение потока всегда.

Теперь сам

1. Слова «he», «she», «hers». Какие вхождения в тексте «ushers»?
2. Сколько шагов по ссылкам неудач в худшем случае сделает автомат на тексте длины 1000?
3. Список запретов меняется раз в час, сообщений 50 в секунду. Сколько сообщений обслуживает одна сборка автомата?