【python 走进NLP】两种高效过滤敏感词算法--DFA算法和AC自动机算法

一道bat面试题：快速替换10亿条标题中的5万个敏感词，有哪些解决思路？
有十亿个标题，存在一个文件中，一行一个标题。有5万个敏感词，存在另一个文件。写一个程序过滤掉所有标题中的所有敏感词，保存到另一个文件中。

1、DFA过滤敏感词算法

在实现文字过滤的算法中，DFA是比较好的实现算法。DFA即Deterministic Finite Automaton，也就是确定有穷自动机。
算法核心是建立了以敏感词为基础的许多敏感词树。

python 实现DFA算法：

# -*- coding:utf-8 -*-import time
time1=time.time()# DFA算法
class DFAFilter():def __init__(self):self.keyword_chains = {}self.delimit = '\x00'def add(self, keyword):keyword = keyword.lower()chars = keyword.strip()if not chars:returnlevel = self.keyword_chainsfor i in range(len(chars)):if chars[i] in level:level = level[chars[i]]else:if not isinstance(level, dict):breakfor j in range(i, len(chars)):level[chars[j]] = {}last_level, last_char = level, chars[j]level = level[chars[j]]last_level[last_char] = {self.delimit: 0}breakif i == len(chars) - 1:level[self.delimit] = 0def parse(self, path):with open(path,encoding='utf-8') as f:for keyword in f:self.add(str(keyword).strip())def filter(self, message, repl="*"):message = message.lower()ret = []start = 0while start < len(message):level = self.keyword_chainsstep_ins = 0for char in message[start:]:if char in level:step_ins += 1if self.delimit not in level[char]:level = level[char]else:ret.append(repl * step_ins)start += step_ins - 1breakelse:ret.append(message[start])breakelse:ret.append(message[start])start += 1return ''.join(ret)if __name__ == "__main__":gfw = DFAFilter()path="F:/文本反垃圾算法/sensitive_words.txt"gfw.parse(path)text="新疆骚乱苹果新品发布会雞八"result = gfw.filter(text)print(text)print(result)time2 = time.time()print('总共耗时：' + str(time2 - time1) + 's')

运行效果：

E:\laidefa\python.exe "E:/Program Files/pycharmproject/敏感词过滤算法/敏感词过滤算法DFA.py"
新疆骚乱苹果新品发布会雞八
****苹果新品发布会**
总共耗时：0.0010344982147216797sProcess finished with exit code 0

2、AC自动机过滤敏感词算法

AC自动机：一个常见的例子就是给出n个单词，再给出一段包含m个字符的文章，让你找出有多少个单词在文章里出现过。
简单地讲，AC自动机就是字典树+kmp算法+失配指针

# -*- coding:utf-8 -*-import time
time1=time.time()# AC自动机算法
class node(object):def __init__(self):self.next = {}self.fail = Noneself.isWord = Falseself.word = ""class ac_automation(object):def __init__(self):self.root = node()# 添加敏感词函数def addword(self, word):temp_root = self.rootfor char in word:if char not in temp_root.next:temp_root.next[char] = node()temp_root = temp_root.next[char]temp_root.isWord = Truetemp_root.word = word# 失败指针函数def make_fail(self):temp_que = []temp_que.append(self.root)while len(temp_que) != 0:temp = temp_que.pop(0)p = Nonefor key,value in temp.next.item():if temp == self.root:temp.next[key].fail = self.rootelse:p = temp.failwhile p is not None:if key in p.next:temp.next[key].fail = p.failbreakp = p.failif p is None:temp.next[key].fail = self.roottemp_que.append(temp.next[key])# 查找敏感词函数def search(self, content):p = self.rootresult = []currentposition = 0while currentposition < len(content):word = content[currentposition]while word in p.next == False and p != self.root:p = p.failif word in p.next:p = p.next[word]else:p = self.rootif p.isWord:result.append(p.word)p = self.rootcurrentposition += 1return result# 加载敏感词库函数def parse(self, path):with open(path,encoding='utf-8') as f:for keyword in f:self.addword(str(keyword).strip())# 敏感词替换函数def words_replace(self, text):""":param ah: AC自动机:param text: 文本:return: 过滤敏感词之后的文本"""result = list(set(self.search(text)))for x in result:m = text.replace(x, '*' * len(x))text = mreturn textif __name__ == '__main__':ah = ac_automation()path='F:/文本反垃圾算法/sensitive_words.txt'ah.parse(path)text1="新疆骚乱苹果新品发布会雞八"text2=ah.words_replace(text1)print(text1)print(text2)time2 = time.time()print('总共耗时：' + str(time2 - time1) + 's')

E:\laidefa\python.exe "E:/Program Files/pycharmproject/敏感词过滤算法/AC自动机过滤敏感词算法.py"
新疆骚乱苹果新品发布会雞八
****苹果新品发布会**
总共耗时：0.0010304450988769531sProcess finished with exit code 0

3、java 实现参考链接：
https://www.cnblogs.com/AlanLee/p/5329555.html

4、敏感词生成

# -*- coding:utf-8 -*-path = 'F:/文本反垃圾算法/sensitive_worlds7.txt'
from 敏感词过滤算法.langconv import *
import pandas as pd
import pypinyin# 文本转拼音
def pinyin(text):""":param text: 文本:return: 文本转拼音"""gap = ' 'piny = gap.join(pypinyin.lazy_pinyin(text))return piny# 繁体转简体
def tradition2simple(text):""":param text: 要过滤的文本:return: 繁体转简体函数"""line = Converter('zh-hans').convert(text)return linedata=pd.read_csv(path,sep='\t')chinise_lable=[]
chinise_type=data['type']for i in data['lable']:line=tradition2simple(i)chinise_lable.append(line)chg_data=pd.DataFrame({'lable':chinise_lable,'type':chinise_type})eng_lable=[]
eng_type=data['type']
for i in data['lable']:# print(i)piny=pinyin(i)# print(piny)eng_lable.append(piny)eng_data=pd.DataFrame({'lable':eng_lable,'type':eng_type})
# print(eng_data)
# 合并
result=chg_data.append(eng_data,ignore_index=True)# 数据框去重res = result.drop_duplicates()
print(res)# 输出
res.to_csv('F:/文本反垃圾算法/中英混合的敏感词10.txt',header=True,index=False,sep='\t',encoding='utf-8')