import re
class BOWInvertedIndexEngine(SearchEngineBase):def __init__(self):super(BOWInvertedIndexEngine, self).__init__()self.inverted_index = dict()#生成索引 word -> iddef process_corpus(self, id, text):words = self.parse_text_to_words(text)for word in words:if word not in self.inverted_index:self.inverted_index[word] = []self.inverted_index[word].append(id) #{'little':['1.txt','2.txt'],...}def search(self, query):query_words = list(self.parse_text_to_words(query))query_words_index = list()for query_word in query_words:query_words_index.append(0)# 如果某一个查询单词的倒序索引为空，我们就立刻返回for query_word in query_words:if query_word not in self.inverted_index:return []result = []while True:# 首先，获得当前状态下所有倒序索引的 indexcurrent_ids = []for idx, query_word in enumerate(query_words):current_index = query_words_index[idx]current_inverted_list = self.inverted_index[query_word] #['1.txt','2.txt']# 已经遍历到了某一个倒序索引的末尾，结束 searchif current_index >= len(current_inverted_list):return resultcurrent_ids.append(current_inverted_list[current_index])# 然后，如果 current_ids 的所有元素都一样，那么表明这个单词在这个元素对应的文档中都出现了if all(x == current_ids[0] for x in current_ids):result.append(current_ids[0])query_words_index = [x + 1 for x in query_words_index]continue# 如果不是，我们就把最小的元素加一min_val = min(current_ids)min_val_pos = current_ids.index(min_val)query_words_index[min_val_pos] += 1@staticmethoddef parse_text_to_words(text):# 使用正则表达式去除标点符号和换行符text = re.sub(r'[^\w ]', ' ', text)# 转为小写text = text.lower()# 生成所有单词的列表word_list = text.split(' ')# 去除空白单词word_list = filter(None, word_list)# 返回单词的 setreturn set(word_list)search_engine = BOWInvertedIndexEngine()
main(search_engine)########## 输出 ########### little
# found 2 result(s):
# 1.txt
# 2.txt
# little vicious
# found 1 result(s):
# 2.txt

LRUCache

import pylru
class LRUCache(object):def __init__(self, size=32):self.cache = pylru.lrucache(size)def has(self, key):return key in self.cachedef get(self, key):return self.cache[key]def set(self, key, value):self.cache[key] = valueclass BOWInvertedIndexEngineWithCache(BOWInvertedIndexEngine, LRUCache):def __init__(self):super(BOWInvertedIndexEngineWithCache, self).__init__()LRUCache.__init__(self)def search(self, query):if self.has(query):print('cache hit!')return self.get(query)result = super(BOWInvertedIndexEngineWithCache, self).search(query)self.set(query, result)return resultsearch_engine = BOWInvertedIndexEngineWithCache()
main(search_engine)########## 输出 ##########
# little
# found 2 result(s):
# 1.txt
# 2.txt
# little
# cache hit!
# found 2 result(s):
# 1.txt
# 2.txt