文本相似度分析实现
内容纲要
文本相似度分析实现
项目结构
文本相似度
文本相似度指的是计算两篇文本的相似度，基于 MinHash算法思想。
调用举例：
text1 = file_util.read_whole_file('data/test001.txt')
text2 = file_util.read_whole_file('data/test002.txt')
similarity = text_similarity.calc_similarity(text1, text2)输出：
data/test001.txt and data/test002.txt similarity: 0.94abstract包
Auto_abstract
import os
from heapq import nlargest
from itertools import count, product
from nlp.utils import nlp_util
class AutoAbstract(object):
def __init__(self, use_stopword=True,
stop_words_file='etc/stopwords.txt',
dict_path=None,
max_iter=100,
tol=0.0001):
if dict_path:
raise RuntimeError("True")
self.__use_stopword = use_stopword
self.__dict_path = dict_path
self.__max_iter = max_iter
self.__tol = tol
self.__stop_words = set()
if stop_words_file:
self.__stop_words_file = stop_words_file
if use_stopword:
for word in open(self.__stop_words_file, 'r', encoding='utf-8'):
self.__stop_words.add(word.strip())
def generate_abstract(self, text_content, sentence_num):
"""
提取文本的主题思想（摘要）
:param text_content: 文本内容
:param sentence_num: 取概率最高的前几句话作为摘要返回
:return: 摘要
"""
text_content = nlp_util.as_text(text_content)
tokens = nlp_util.cut_sentences(text_content)
sentences, sents = nlp_util.cut_filter_words(tokens, self.__stop_words, self.__use_stopword)
graph = self.create_graph(sents)
scores = nlp_util.weight_map_rank(graph, self.__max_iter, self.__tol)
sent_selected = nlargest(sentence_num, zip(scores, count()))
if sentence_num > len(sent_selected):
sentence_num = len(sent_selected)
sent_index = []
for i in range(sentence_num):
sent_index.append(sent_selected[i][1])
result_sent = [sentences[i] for i in sent_index]
return os.linesep.join(result_sent).replace(' ', '')
@staticmethod
def create_graph(word_sent):
num = len(word_sent)
board = [[0.0 for _ in range(num)] for _ in range(num)]
for i, j in product(range(num), repeat=2):
if i != j:
board[i][j] = nlp_util.sentences_similarity(word_sent[i], word_sent[j])
return board
nroute
import re
from math import log
from nlp.abstract.perceptron import Perceptron
re_eng = re.compile('[a-zA-Z0-9]', re.U)
re_han = re.compile("([\u4E00-\u9FD5a-zA-Z0-9+#&\._%\-]+)", re.U)
re_skip = re.compile("(\r\n|\s)", re.U)
class Segment:
def __init__(self):
self.vocab = {}
self.max_word_len = 0
self.max_freq = 0
self.total_freq = 0
self.initialized = False
self.model = None
def init(self, vocab_path='data/abstract/dict/jiagu.dict', user_vocab='data/abstract/dict/user.dict',
model_path='data/abstract/model/cws.model'):
self.load_vocab(vocab_path)
self.load_vocab(user_vocab)
self.model = Perceptron(model_path)
self.initialized = True
def load_vocab(self, vocab_path):
fin = open(vocab_path, 'r', encoding='utf8')
for index, line in enumerate(fin):
line = line.strip()
if line == '':
continue
word_freq_tag = line.split('\t')
if len(word_freq_tag) == 1:
word = word_freq_tag[0]
self.add_vocab(word)
elif len(word_freq_tag) == 2:
word = word_freq_tag[0]
freq = int(word_freq_tag[1])
self.add_vocab(word, freq)
fin.close()
def add_vocab(self, word=None, freq=None, tag=None):
if freq == None:
freq = self.max_freq
if word not in self.vocab:
self.vocab[word] = 0
self.vocab[word] += freq
self.total_freq += freq
if freq > self.max_freq:
self.max_freq = freq
if len(word) > self.max_word_len:
self.max_word_len = len(word)
def del_vocab(self, word=None, freq=None, tag=None):
if word not in self.vocab:
return None
vocab_freq = self.vocab[word]
if freq == None or vocab_freq <= freq:
del self.vocab[word]
self.total_freq -= vocab_freq
else:
self.vocab[word] -= freq
# self.max_freq and self.max_word_len ?
def load_userdict(self, userdict):
if self.initialized == False:
self.init()
if isinstance(userdict, str):
self.load_vocab(userdict)
for item in userdict:
if isinstance(item, list):
if len(item) == 1:
word = item[0]
self.add_vocab(word)
elif len(item) == 2:
word = item[0]
freq = item[1]
self.add_vocab(word, freq)
elif isinstance(item, str):
self.add_vocab(word=item)
def del_userdict(self, userdict):
if self.initialized == False:
self.init()
for item in userdict:
if isinstance(item, list):
if len(item) == 1:
word = item[0]
self.del_vocab(word)
elif len(item) == 2:
word = item[0]
freq = item[1]
self.del_vocab(word, freq)
elif isinstance(item, str):
self.del_vocab(word=item)
def calc_route(self, sentence, DAG, route):
vocab = self.vocab
N = len(sentence)
route[N] = (0, 0)
logtotal = log(self.total_freq)
for idx in range(N - 1, -1, -1):
route[idx] = max(
(log(vocab.get(sentence[idx:x + 1]) or 1) - logtotal + route[x + 1][0], x) for x in DAG[idx])
def create_DAG(self, sentence):
vocab = self.vocab
max_word_len = self.max_word_len
DAG = {}
N = len(sentence)
for idx in range(N):
cand_idx = [idx]
for i in range(idx + 1, idx + min(max_word_len, N - idx), 1):
cand = sentence[idx: i + 1]
if cand in vocab:
cand_idx.append(i)
DAG[idx] = cand_idx
return DAG
def cut_search(self, sentence):
DAG = self.create_DAG(sentence)
old_j = -1
for k, L in DAG.items():
if len(L) == 1 and k > old_j:
yield sentence[k:L[0] + 1]
old_j = L[0]
else:
for j in L:
if j > k:
yield sentence[k:j + 1]
old_j = j
def cut_vocab(self, sentence):
DAG = self.create_DAG(sentence)
route = {}
self.calc_route(sentence, DAG, route)
x = 0
N = len(sentence)
buf = ''
while x < N:
y = route[x][1] + 1
l_word = sentence[x:y]
if buf:
yield buf
buf = ''
yield l_word
x = y
if buf:
yield buf
buf = ''
def cut_words(self, sentence):
DAG = self.create_DAG(sentence)
route = {}
self.calc_route(sentence, DAG, route)
x = 0
N = len(sentence)
buf = ''
while x < N:
y = route[x][1] + 1
l_word = sentence[x:y]
if re_eng.match(l_word) and len(l_word) == 1:
buf += l_word
x = y
else:
if buf:
yield buf
buf = ''
yield l_word
x = y
if buf:
yield buf
buf = ''
def model_cut(self, sentence):
if sentence == '':
return ['']
sentence = list(sentence)
labels = self.model.predict(sentence)
return self.__lab2word(sentence, labels)
def __lab2word(self, sentence, labels):
sen_len = len(sentence)
tmp_word = ""
words = []
for i in range(sen_len):
label = labels[i]
w = sentence[i]
if label == "B":
tmp_word += w
elif label == "M":
tmp_word += w
elif label == "E":
tmp_word += w
words.append(tmp_word)
tmp_word = ""
else:
if tmp_word != '':
words.append(tmp_word)
tmp_word = ""
words.append(w)
if tmp_word:
words.append(tmp_word)
return words
def seg_default(self, sentence):
blocks = re_han.split(sentence)
cut_block = self.cut_words
cut_all = False
for block in blocks:
if not block:
continue
if re_han.match(block):
for word in cut_block(block):
yield word
else:
tmp = re_skip.split(block)
for x in tmp:
if re_skip.match(x):
yield x
elif not cut_all:
for xx in x:
yield xx
else:
yield x
def seg_new_word(self, sentence):
blocks = re_han.split(sentence)
cut_block = self.cut_words
cut_all = False
for block in blocks:
if not block:
continue
if re_han.match(block):
words1 = list(cut_block(block))
# print(words1)
words2 = self.model_cut(block)
# print(words2)
new_word = [] # 有冲突的不加，长度大于4的不加，加完记得删除
length = len(words1)
for n in range(3):
can_limit = length - n + 1
for i in range(0, can_limit):
ngram = ''.join(words1[i:i + n])
word_len = len(ngram)
if word_len > 4 or word_len == 1:
continue
if ngram in words2 and ngram not in words1:
# print(ngram)
new_word.append([ngram, 1])
# new_word = []
# for word in words2:
# if word not in words1 and len(word)>1 and len(word) < 4 :#and not re_eng.match(word):
# new_word.append([word, 1])
self.load_userdict(new_word)
# print('------------------')
for word in cut_block(block):
yield word
# 删除字典
self.del_userdict(new_word)
else:
tmp = re_skip.split(block)
for x in tmp:
if re_skip.match(x):
yield x
elif not cut_all:
for xx in x:
yield xx
else:
yield x
def seg(self, sentence, mode="default"):
if self.initialized == False:
self.init()
if mode == 'probe':
return list(self.seg_new_word(sentence))
else:
return list(self.seg_default(sentence))
perceptron
# -*- coding:utf-8 -*-
import gzip
import pickle
import random
from collections import defaultdict
class AveragedPerceptron(object):
def __init__(self):
# Each feature gets its own weight vector, so weights is a dict-of-dicts
self.weights = {}
self.classes = set()
# The accumulated values, for the averaging. These will be keyed by
# feature/clas tuples
self._totals = defaultdict(int)
# The last time the feature was changed, for the averaging. Also
# keyed by feature/clas tuples
# (tstamps is short for timestamps)
self._tstamps = defaultdict(int)
# Number of instances seen
self.i = 0
def predict(self, features):
'''Dot-product the features and current weights and return the best label.'''
scores = defaultdict(float)
for feat, value in features.items():
if feat not in self.weights or value == 0:
continue
weights = self.weights[feat]
for label, weight in weights.items():
scores[label] += value * weight
# Do a secondary alphabetic sort, for stability
return max(self.classes, key=lambda label: (scores[label], label))
def update(self, truth, guess, features):
'''Update the feature weights.'''
def upd_feat(c, f, w, v):
param = (f, c)
self._totals[param] += (self.i - self._tstamps[param]) * w
self._tstamps[param] = self.i
self.weights[f][c] = w + v
self.i += 1
if truth == guess:
return None
for f in features:
weights = self.weights.setdefault(f, {})
upd_feat(truth, f, weights.get(truth, 0.0), 1.0)
upd_feat(guess, f, weights.get(guess, 0.0), -1.0)
return None
def average_weights(self):
'''Average weights from all iterations.'''
for feat, weights in self.weights.items():
new_feat_weights = {}
for clas, weight in weights.items():
param = (feat, clas)
total = self._totals[param]
total += (self.i - self._tstamps[param]) * weight
averaged = round(total / float(self.i), 3)
if averaged:
new_feat_weights[clas] = averaged
self.weights[feat] = new_feat_weights
return None
class Perceptron:
def __init__(self, loc=None):
self.START = ['-START-', '-START2-']
self.END = ['-END-', '-END2-']
self.model = AveragedPerceptron()
if loc != None:
self.load(loc)
def predict(self, words):
prev, prev2 = self.START
labels = []
context = self.START + words + self.END
for i, word in enumerate(words):
features = self._get_features(i, word, context, prev, prev2)
tag = self.model.predict(features)
labels.append(tag)
prev2 = prev
prev = tag
return labels
def train(self, sentences, save_loc=None, nr_iter=5, shuf=False):
self._make_tagdict(sentences)
for iter_ in range(nr_iter):
c = 0
n = 0
for words, tags in sentences:
prev, prev2 = self.START
context = self.START + words + self.END
for i, word in enumerate(words):
feats = self._get_features(i, word, context, prev, prev2)
guess = self.model.predict(feats)
self.model.update(tags[i], guess, feats)
prev2 = prev
prev = guess
c += guess == tags[i]
n += 1
if shuf == True:
random.shuffle(sentences)
print("Iter {0}: {1}/{2}={3}".format(iter_, c, n, (float(c) / n) * 100))
self.save(save_loc)
self.model.average_weights()
self.save(save_loc)
def save(self, loc='model/ap.model', zip=True):
if zip == False:
pickle.dump((self.model.weights, self.model.classes), open(loc, 'wb'))
else:
pickle.dump((self.model.weights, self.model.classes), gzip.open(loc, 'wb'))
def load(self, loc='model/ap.model', zip=True):
if zip == False:
self.model.weights, self.model.classes = pickle.load(open(loc, 'rb'))
else:
self.model.weights, self.model.classes = pickle.load(gzip.open(loc, 'rb'))
def _get_features(self, i, word, context, prev, prev2):
'''Map tokens into a feature representation, implemented as a
{hashable: float} dict. If the features change, a new model must be
trained.
'''
def add(name, *args):
features[' '.join((name,) + tuple(args))] += 1
i += len(self.START)
features = defaultdict(int)
# It's useful to have a constant feature, which acts sort of like a prior
add('bias')
add('i suffix', word[-3:])
add('i pref1', word[0])
add('i-1 tag', prev)
add('i-2 tag', prev2)
add('i tag+i-2 tag', prev, prev2)
add('i word', context[i])
add('i-1 tag+i word', prev, context[i])
add('i-1 word', context[i - 1])
add('i-1 suffix', context[i - 1][-3:])
add('i-2 word', context[i - 2])
add('i+1 word', context[i + 1])
add('i+1 suffix', context[i + 1][-3:])
add('i+2 word', context[i + 2])
return features
def _make_tagdict(self, sentences):
'''Make a tag dictionary for single-tag words.'''
for words, tags in sentences:
for word, tag in zip(words, tags):
self.model.classes.add(tag)
def train(filepath='data/train.txt', model='model/ap.model', nr_iter=1):
tagger = Perceptron()
print('Reading corpus...')
training_data = []
sentence = ([], [])
fin = open(filepath, 'r', encoding='utf8')
for index, line in enumerate(fin):
line = line.strip()
if line == '':
training_data.append(sentence)
sentence = ([], [])
else:
params = line.split()
if len(params) != 2: continue
sentence[0].append(params[0])
sentence[1].append(params[1])
fin.close()
print('training corpus size : %d', len(training_data))
print('Start training...')
tagger.train(training_data, save_loc=model, nr_iter=nr_iter)
def eval(filepath='data/test.txt', model='model/ap.model'):
tagger = Perceptron(model)
print('Start testing...')
right = 0.0
total = 0.0
sentence = ([], [])
fin = open(filepath, 'r', encoding='utf8')
for index, line in enumerate(fin):
line = line.strip()
if line == '':
words = sentence[0]
tags = sentence[1]
outputs = tagger.predict(words)
assert len(tags) == len(outputs)
total += len(tags)
for o, t in zip(outputs, tags):
if o == t: right += 1
sentence = ([], [])
else:
params = line.split()
if len(params) != 2: continue
sentence[0].append(params[0])
sentence[1].append(params[1])
fin.close()
print("Precision : %f", right / total)
def predict(model='model/ap.model'):
tagger = Perceptron(model)
while True:
text = input('>')
words = list(text)
labels = tagger.predict(words)
for word, label in zip(words, labels):
print(word, label)
if __name__ == '__main__':
train()
eval()
# predict()
similarity包
Text_similarity
import nltk
from datasketch import MinHash
ngrams_num = 3
def calc_similarity(text1, text2):
"""
计算两段文本的相似度
:param text1:
:param text2:
:return:
"""
# print(text1)
# print(text2)
m1, m2 = MinHash(), MinHash()
for d in nltk.ngrams(text1, ngrams_num):
m1.update("".join(d).encode('utf8'))
for d in nltk.ngrams(text2, ngrams_num):
m2.update("".join(d).encode('utf8'))
s1 = set(text1)
s2 = set(text2)
sty = float(len(s1.intersection(s2))) / float(len(s1.union(s2)))
return round(sty, 2)
utils包
File_util
import hashlib
import os
import traceback
import chardet
DEFAULT_ENCODING = 'utf-8'
def write_lines(lines, file_name):
"""
把字符串数组写入到指定文件里
:param lines:
:param file_name:
:return:
"""
with open(file_name, 'w', encoding=DEFAULT_ENCODING) as text_file:
text_file.write('\n'.join(lines))
def append_line(line, file_name):
"""
往指定的文件追加一行
:param line:
:param file_name:
:return:
"""
with open(file_name, 'a', encoding=DEFAULT_ENCODING) as text_file:
text_file.write(line + '\n')
def append_lines(lines, file_name):
"""
往指定的文件追加多行
:param lines:
:param file_name:
:return:
"""
with open(file_name, 'a', encoding=DEFAULT_ENCODING) as text_file:
text_file.write('\n'.join(lines))
def read_whole_file(file_name):
"""
读取文本文件的所有内容
:param file_name:
:return:
"""
if not os.path.exists(file_name):
return ''
with open(file_name, 'rb') as text_file:
content = text_file.read()
encoding = chardet.detect(content)['encoding']
return content.decode(encoding)
def read_lines(file_name):
"""
按行读取文本文件并返回
:param file_name:
:return:
"""
content = read_whole_file(file_name).replace('\r', '')
return content.split('\n')
def calc_file_md5(file_name):
with open(file_name, 'rb') as reader:
content = reader.read()
return hashlib.md5(content).hexdigest()
def get_text_file_encoding(file_name):
if not os.path.exists(file_name):
return ''
with open(file_name, 'rb') as text_file:
content = text_file.read()
return chardet.detect(content)['encoding']
def move_file_2_dic(source_file, dest_dic):
"""
移动文件到目标文件夹，若目标文件夹下有同名文件直接覆盖
:param source_file:
:param dest_dic:
:return:
"""
target_file = os.path.join(dest_dic, source_file[source_file.rfind('/') + 1:])
if os.path.exists(target_file):
os.remove(target_file)
os.rename(source_file, target_file)
def get_file_suffix(file_name):
"""
获取文件后缀
:param file_name:
:return:
"""
if file_name is None:
return ''
strs = file_name.split('.')
if len(strs) > 1:
return strs[len(strs) - 1]
return ''
def delete_file(file_path):
"""
删除文件（不会抛异常）
:param file_path:
:return:
"""
try:
if os.path.exists(file_path):
os.remove(file_path)
except:
print('delete_file: {} error!! {}'.format(file_path, traceback.format_exc()))
Nlp_util
import codecs
import math
from jieba import posseg
from nlp.abstract.nroute import Segment
def different(scores, old_scores, tol=0.0001):
flag = False
for i in range(len(scores)):
if math.fabs(scores[i] - old_scores[i]) >= tol: # 原始是0.0001
flag = True
break
return flag
def sentences_similarity(s1, s2):
"""计算两个句子的相似度
:param s1: list
:param s2: list
:return: float
"""
counter = 0
for sent in s1:
if sent in s2:
counter += 1
if counter == 0:
return 0
return counter / (math.log(len(s1) + len(s2)))
def cut_words(text, stop_flag=['x', 'c', 'u', 'd', 'p', 't', 'uj', 'm', 'f', 'r']):
""" 将文本切成词语 """
stop_words = codecs.open('etc/stopwords.txt', 'r', encoding='utf8').readlines()
stop_words = [w.strip() for w in stop_words]
result = []
words = posseg.cut(text)
for word, flag in words:
if flag not in stop_flag and word not in stop_words:
result.append(word)
return result
def cut_sentences(text):
"""
将文本切成句子
:param text: 待切分文本
:return: 句子列表
"""
text = text.translate(str.maketrans('!?！？\r\n', '。。。。 '))
return [sen for sen in text.split('。') if len(sen.strip()) > 1]
def as_text(v):
"""生成unicode字符串"""
if v is None:
return None
elif isinstance(v, bytes):
return v.decode('utf-8', errors='ignore')
elif isinstance(v, str):
return v
else:
raise ValueError('Unknown type %r' % type(v))
def cut_filter_words(cutted_sentences, stopwords, use_stopwords=False):
seg = Segment()
sentences = []
sents = []
for sent in cutted_sentences:
sentences.append(sent)
if use_stopwords:
sents.append([word for word in seg.seg(sent) if word and word not in stopwords]) # 把句子分成词语
else:
sents.append([word for word in seg.seg(sent) if word])
return sentences, sents
def weight_map_rank(weight_graph, max_iter, tol):
# 初始分数设置为0.5
# 初始化每个句子的分子和老分数
scores = [0.5 for _ in range(len(weight_graph))]
old_scores = [0.0 for _ in range(len(weight_graph))]
denominator = get_degree(weight_graph)
# 开始迭代
count = 0
while different(scores, old_scores, tol):
for i in range(len(weight_graph)):
old_scores[i] = scores[i]
# 计算每个句子的分数
for i in range(len(weight_graph)):
scores[i] = get_score(weight_graph, denominator, i)
count += 1
if count > max_iter:
break
return scores
def get_degree(weight_graph):
length = len(weight_graph)
denominator = [0.0 for _ in range(len(weight_graph))]
for j in range(length):
for k in range(length):
denominator[j] += weight_graph[j][k]
if denominator[j] == 0:
denominator[j] = 1.0
return denominator
def get_score(weight_graph, denominator, i):
"""
:param weight_graph:
:param denominator:
:param i: int
第i个句子
:return: float
"""
length = len(weight_graph)
d = 0.85
added_score = 0.0
for j in range(length):
# [j,i]是指句子j指向句子i
fraction = weight_graph[j][i] * 1.0
# 除以j的出度
added_score += fraction / denominator[j]
weighted_score = (1 - d) + d * added_score
return weighted_score
测试
from nlp.similarity import text_similarity
from nlp.utils import file_util
def text_similarity_demo():
""" 文本相似度 """
file1 = '/Users/davidlee/Desktop/python/sentence_demo02/data/test001.txt'
file2 = '/Users/davidlee/Desktop/python/sentence_demo02/data/test002.txt'
file3 = '/Users/davidlee/Desktop/python/sentence_demo02/data/test005.txt'
text1 = file_util.read_whole_file(file1)
text2 = file_util.read_whole_file(file2)
text3 = file_util.read_whole_file(file3)
sty1 = text_similarity.calc_similarity(text1, text2)
sty2 = text_similarity.calc_similarity(text1, text3)
print("{} and {} similarity: {}".format(file1, file2, sty1))
print("{} and {} similarity: {}".format(file1, file3, sty2))
print('--' * 55)
if __name__ == '__main__':
text_similarity_demo()
requirments
nltk~=3.6.7
setuptools~=28.8.0
datasketch~=1.5.9
chardet~=5.0.0
jieba~=0.42.1setup
from distutils.core import setup
from setuptools import find_packages
LONGDOC = """
nlp tools
"""
setup(name='sentence_demo1',
version='1.0.0',
description='sentence_demo1',
long_description=LONGDOC,
author='Li',
url='https://xn--rgv085f.tech/',
classifiers=[
'Intended Audience :: Developers',
'Operating System :: OS Independent',
'Natural Language :: Chinese (Simplified)',
'Natural Language :: Chinese (Traditional)',
'Programming Language :: Python',
'Programming Language :: Python :: 3.6',
'Topic :: sentence_demo1'
],
keywords='sentence',
packages=find_packages('sentence'),
package_dir={'': 'sentence'},
package_data={'sentence': ['*.*', '*/*']}
)