Python实战之手写一个搜索引擎

def get_entry(self):
    baseurl = 'Http://his.CSSn.cn/lsx/sjls/'
    entries = []
    for idx in range(5):
        entry = baseurl if idx == 0 else baseurl + 'index_' + str(idx) + '.shtml'
        entries.append(entry)
    return entries

def parse4links(self, entries):
    links = []
    headers = {
        'USER-AGENT': 'Mozilla/5.0 (windows NT 10.0; WOW64) AppleWEBKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36'
    }
    for entry in entries:
        try:
            response = requests.get(url = entry, headers = headers)
            html = response.text.encode(response.encoding).decode('utf-8')
            time.sleep(0.5)
        except:
            continue

        html_parser = etree.HTML(html)
        link = html_parser.xpath('//div[@class="ImageListView"]/ol/li/a/@href')
        link_filtered = [url for url in link if 'www' not in url]
        link_complete = [entry + url.lstrip('./') for url in link_filtered]
        links.extend(link_complete)

    return links

def parse4datas(self, entries):
    datas = []
    headers = {
        'USER-AGENT': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36'
    }
    data_count = 0
    for entry in entries:
        try:
            response = requests.get(url = entry, headers = headers)
            html = response.text.encode(response.encoding).decode('utf-8')
            time.sleep(0.2)
        except:
            continue

        html_parser = etree.HTML(html)
        title = html_parser.xpath('//span[@class="TitleFont"]/text()')
        content = html_parser.xpath('//div[@class="TRS_Editor"]//p//text()')
        content = [cont.replace('\u3000', '').replace('\xa0', '').replace('\n', '').replace('\t', '') for cont in content]
        content = [cont for cont in content if len(cont) > 30 and not re.search(r'[《|》]', cont)]

        if len(title) != 0 or len(content) != 0:
            data_count += 1
            datas.append({
                'id'  : data_count,
                'link': entry,
                'cont': '\t'.join(content),
                'title': title[0]
            })

    return datas

def grab_data(self):
    # 获取入口链接
    entries = self.get_entry()
    # 遍历入口链接，解析得到文章链接
    links = self.parse4links(entries)
    # 遍历文章链接，解析得到文章内容
    datas = self.parse4datas(links)
    # 将相关数据写入变量 data
    self.data = pd.DataFrame(datas)

def save_data(self):
    # 将变量 data 写入 csv 文件
    self.data.to_csv(self.data_path, index = None)

|---------------------------------------------------|
|    id    |     link   |     cont     |    title   |
|---------------------------------------------------|
|  page id |  page link | page content | page title |
|---------------------------------------------------|
|  ......  |   ......   |    ......    |   ......   |
|---------------------------------------------------|

def make_iindex(self):
    # 读取数据
    df = pd.read_csv(self.data_path)
    # 特殊变量，用于搜索模块
    TOTAL_DOC_NUM = 0 # 总文章数量
    TOTAL_DOC_LEN = 0 # 总文章长度
    # 遍历每一行
    for row in df.itertuples():
        doc_id = getattr(row, 'id') # 文章标识
        cont = getattr(row, 'cont') # 文章内容

        TOTAL_DOC_NUM += 1
        TOTAL_DOC_LEN += len(cont)

        # 对文章内容分词
        # 并将其变成 {Word: frequency, ...} 的形式
        cuts = jieba.lcut_for_search(cont)
        word2freq = self.fORMat(cuts)

        # 遍历每个词，将相关数据写入变量 iindex
        for word in word2freq:
            meta = {
                'id': doc_id,
                'dl': len(word2freq),
                'tf': word2freq[word]
            }
            if word in self.iindex:
                self.iindex[word]['df'] = self.iindex[word]['df'] + 1
                self.iindex[word]['ds'].append(meta)
            else:
                self.iindex[word] = {}
                self.iindex[word]['df'] = 1
                self.iindex[word]['ds'] = []
                self.iindex[word]['ds'].append(meta)

    # 将特殊变量写入配置文件
    self.config.set('DATA', 'TOTAL_DOC_NUM', str(TOTAL_DOC_NUM)) # 文章总数
    self.config.set('DATA', 'AVG_DOC_LEN', str(TOTAL_DOC_LEN / TOTAL_DOC_NUM)) # 文章平均长度
    with open(self.option['filepath'], 'w', encoding = self.option['encoding']) as config_file:
        self.config.write(config_file)

def save_iindex(self):
    # 将变量 iindex 写入 JSON 文件
    fd = open(self.iindex_path, 'w', encoding = 'utf-8')
    json.dump(self.iindex, fd, ensure_ascii = False)
    fd.close()

{
    word: {
        'df': document_frequency,
        'ds': [{
            'id': document_id,
            'dl': document_length,
            'tf': term_frequency
        }, ...]
    },
    ...
}

def search(self, query):
    BM25_scores = {}

    # 对用户输入分词
    # 并将其变成 {word: frequency, ...} 的形式
    query = jieba.lcut_for_search(query)
    word2freq = self.format(query)

    # 遍历每个词
    # 计算每个词与相关文章之间的得分（计算公式参考 BM25 算法）
    for word in word2freq:
        data = self.iindex.get(word)
        if not data:
            continue
        BM25_score = 0
        qf = word2freq[word]
        df = data['df']
        ds = data['ds']
        W = math.log((self.N - df + 0.5) / (df + 0.5))
        for doc in ds:
            doc_id = doc['id']
            tf = doc['tf']
            dl = doc['dl']
            K = self.k1 * (1 - self.b + self.b * (dl / self.AVGDL))
            R = (tf * (self.k1 + 1) / (tf + K)) * (qf * (self.k2 + 1) / (qf + self.k2))
            BM25_score = W * R
            BM25_scores[doc_id] = BM25_scores[doc_id] + BM25_score if doc_id in BM25_scores else BM25_score

    # 对所有得分按从大到小的顺序排列，返回结果
    BM25_scores = sorted(BM25_scores.items(), key = lambda item: item[1])
    BM25_scores.reverse()
    return BM25_scores