基于Python实现的通用小规模搜索引擎
1.项目简介
1.1背景
《信息内容安全》网络信息内容获取技术课程项目设计
- 一个至少能支持10个以上网站的爬虫程序，且支持增量式数据采集;并至少采集10000个实际网页;
- 针对采集回来的网页内容， 能够实现网页文本的分类;
- 可进行重复或冗余网页的去重过滤;
- 对经去冗以后的内容建立倒排索引;
- 采用PageRank算法实现搜索结果的排序;
- 支持自然语言的模糊检索;
- 可实现搜索结果的可视化呈现。
- 可以在线记录每次检索的日志，井可对日志数据进统计分析和关联挖掘。
1.2运行环境
- 平台：全平台
- jdk 1.8.0
- ElasticSearch 7.4.0
- Python 3.6 及以上
-
- 安装依赖模块
-
-
- PageRank算法、AI文本分类与上传
-
> pip install paddlepaddle numpy elasticsearch-
-
- 数据的爬取与预处理
-
> pip install requests bs41.3运行步骤
安装配置ElasticSearch并启动
- 下载 并解压Elasticsearch，详细步骤自行搜索
-
- 可以从 apt 和 yum 的软件仓库安装，也可以使用 Windows MSI 安装包安装
- 安装 IK 中文分词器，详细步骤自行搜索
- 创建索引
PUT http://127.0.0.1/page
{
"settings": {
"number_of_shards": "5",
"number_of_replicas": "0"
},
"mappings": {
"properties": {
"title": {
"type": "text",
"analyzer": "ik_max_word"
},
"weight": {
"type": "double"
},
"content" : {
"type" : "text",
"analyzer": "ik_max_word"
},
"content_type": {
"type": "text"
},
"url": {
"type": "text",
"analyzer": "ik_max_word"
},
"update_date": {
"type": "date",
"format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
}
}
}
}- 启动 ElasticSearch ，在 bash 中执行
bin/elasticsearch或者在 Windows 的 cmd、powershell 执行bin\elasticsearch.bat
启动Web服务
> cd WebApp
> java -jar *.jar
数据的爬取与预处理
> cd DataCrawler
> python crawler.py
计算PageRank值
> cd DataProcess
> python PageRank.py
利用AI进行文本分类并上传至ES
> cd DataProcess/Text_Classification
> python Classify.py
2.需求分析
2.1数据描述
2.1.1 静态数据
| 变量名 | 描述 |
|
| 线程个数 |
|
| 种子页面 |
2.1.2 动态数据
| 变量名 | 描述 | 类型 |
|
| 限定域名 | 列表 |
|
| 禁止域名 | 列表 |
|
| 线程个数 | 整型 |
|
| 限定页面个数 | 整型 |
2.1.3索引数据字典
页面（page）信息索引：
| 数据项名称 | 含义 | 别名 | 类型 | 备注 |
|
| 网站标题 |
| 使用 分词 | |
|
| PageRank值 | pr值，PR值 |
| |
|
| 网站中的内容 |
| 使用 分词 | |
|
| 网站中的内容分类 |
| 文化, 娱乐, 体育, 财经, 房产, 汽车, 教育, 科技, 国际, 证券 | |
|
| 网站的链接 |
| 使用 分词 | |
|
| 数据更新的时间 |
|
|| || |
2.2. 数据采集
种子 url 数据从 init_url 列表中选取，并按照顺序，依次以各个 url 为起点进行递归的数据采集
爬取数据的url需要限制在 restricted_url 列表里面
2.3功能需求
2.3.1 数据爬取与预处理功能
利用Python爬虫，执行以下步骤：
- 开始
- 选取一个链接作为起点
- 如果爬取的网页总数达到要求，则结束，否则执行第 4 步
- 爬取指定链接的相关信息，并获取当前网站中的所有链接
- 对 4 中获取的网站中的所有链接中的每一条数据，执行过程3
爬取网站如下信息，详细描述见3.4. 索引数据字典
- title
- content
- content_type
- update_date
- url
- link（当前网站中包含的所有链接，用于计算pr值）
2.3.2. 计算 PageRank 功能
根据link计算爬取下来的每个网站的PageRank值，迭代次数为50次。解决pr值呈周期性变化的问题。将pr值作为网站重要程度的指标，并补充到网站信息中
2.3.3. AI 文本分类并提交到 ES 功能
利用深度学习，分析每个页面的content的类别。将类别补充到网站信息中，同时删除网站信息中不再使用的link项，形成最终数据（参考3.4. 索引数据字典），并上传至ES，供用户交互功能调用。
2.3.4. 用户交互功能
设计WebApp，用户通过浏览器访问页面。用户提交搜索信息后，判断合法性，不合法则返回ERROR界面提示用户。如果合法，则后端代码从本地 ES 中查询数据，处理后将结果分条显示到前端。同时通过限制单个ip每分钟的访问次数来简单防御用户恶意搜索。
2.4. 性能需求
2.4.1. 数据精确度
对数据精确度要求不高，主要数据为：
| 项目 | 限制 |
| 爬取的数据总量 | 每小时查询一下数据总量 |
| 查询结果数量 | 匹配的所有结果数 |
| 数据更新日期 | 精确到分钟即可 |
2.4.2. 时间特性
| 项目 | 限制 |
| 每爬取 1 万个网页耗时 | 30 分钟以内 |
| 计算 1 万个网页的pr值耗时 | 10 分钟以内 |
| 对 1 万个网页内容进行AI 进行文本分类并上传至ES耗时 | 10 分钟以内 |
| Web 首页打开耗时 | 5 秒以内 |
| 查询结果页面打开耗时 | 5 秒以内 |
2.5. 运行需求
2.5.1. 用户界面
用户通过浏览器访问，有两个页面，一个是主页，只有简单的输入框提供用户搜索；另一个是一般界面，提供高级搜索功能，并显示搜索结果。
2.5.2. 主页
| 控件 | 作用 | 布局 |
| 图标 | 显示Logo | 居中 |
|输入框|接收用户输入的关键字|Logo图标下偏左
|按钮|提交用户输入的关键字，并返回搜索结果|输入框右|
2.5.3. 搜索结果界面
该界面分为三个部分，导航条、搜索结果、信息展示。这三个部分布局如下
| 部分 | 位置 | height | width |
| 导航条 | 顶部 | 50px | 100% |
| 搜索结果 | 导航条左下部 | auto | 70% |
| 信息展示 | 导航条右下部 | auto | 30% |
导航条部分
以下控件从左向右依次（顺序可以任意）在导航条中排列
| 控件 | 作用 |
| 输入框 | 接收用户输入的关键字 |
| 输入框 | 可以输入域名，将搜索结果限制在该域名内 |
| 数字输入框 | 查询结果分页显示，该框指示跳转到指定的搜索结果页 |
| 选择框 | 允许用户选择匹配方式：标题和内容（默认）、仅标题、仅内容 |
| 选择框 | 选择搜索结果的排序方式：倒排索引（默认）、 PageRank 排序 |
| 按钮 | 提交用户输入的所有数据，并返回搜索结果 |
搜索结果部分
将搜索结果以list的形式展示出来，每个list item显示匹配的网站的如下数据
- 标题
- 内容
- url
- 类别
- PageRank值
- 更新时间
在list结尾，显示分页组件，使用户可以点击跳转，样式如下：
| < | 1 | 2 | 3 | 4 | 5 | 6 | > |
信息展示部分
展示一些必要信息，如：
- 本次查询耗时
- 查询结果数
- 数据库中的数据总数
- 等等
2.5.4 软件接口
| 接口名 | 描述 | 所在模块 | 调用方式 |
|
| 初次启动调用此接口 |
| 内部调用 |
|
| 得到目标 url 的页面 |
| 内部调用 |
|
| 爬虫线程 |
| 内部调用 |
|
| 主任务执行线程 |
|
|
|
| 去掉所有未在 url 中出现的 link 及错误文件 |
| 内部调用 |
|
| 计算PageRank |
| 内部调用 |
|
| 程序运行方法 |
|
|
|
| 获取已爬取数据 |
| 内部调用 |
|
| 利用AI进行文本分类 |
|
|
2.5.5. 故障处理
各个功能模块如果出问题，会出现以下情况：
| 模块 | 出故障后 | 简单排查 |
| 爬虫 | 数据不再更新 | 检查网络，检查内存资源是否不足 |
| PageRank计算 | 数据不再更新 | 检查内存资源和CPU资源是否不足 |
| AI 文本分类 | 数据不再更新 | 检查内存资源和CPU资源是否不足 |
| ElasticSearch | 前端无法获取查询结果 | 问题比较复杂 |
| WebApp | 无法访问网站 | 问题比较复杂 |
其中，后两个模块出问题会造成严重问题，如果重启不能解决问题的话，采用如下措施
| 模块 | 故障排除 | 终极方法 |
| ElasticSearch | ①java环境是否正确 | 在其他机器上部署，并修改WebApp使其到该机器上获取服务 |
| WebApp | ①端口是否被占用 | 在其他机器上部署，并修改域名解析，将域名解析到新机器上 |
2.6. 其他需求
2.6.1. 可维护性
- 网络爬虫设置了黑名单和白名单，可以限制爬取的范围。
- 各个功能分离开，协同工作。同时，只要不修改数据格式，各个模块的修改不会影响其他模块
2.6.2. 可移植性
- WebApp 使用 Spring boot 框架开发，打包后只有一个jar包，可以在任何有java环境的机器上部署
- 其他功能都用python实现，可以部署在任何有python环境的机器上
- ElasticSearch 支持分布式部署，可以部署在任意平台
2.6.3. 数据完整性
- ElasticSearch 支持分布式，会自动将数据备份在不同节点。如果某个节点出了故障，不会破坏数据，也不会影响程序的查询结果
3.代码展示
import os
import sys
import json
import numpy as np
import time
import codecs
dir_path = os.path.split(os.path.realpath(sys.argv[0]))[0] + '/../RawData'
print(dir_path)
Vexname = list(os.listdir(dir_path))
Vexnum = len(Vexname)
epoch = 50
# 初始化，去掉所有未在url中出现的link以及错误文件
def init():
global Vexnum
falsefiles={}
idx=0
start = time.perf_counter()
for file in Vexname:
if idx % 100 == 0:
a = '=' * int(idx / Vexnum * 100)
b = ' ' * (100 - int(idx / Vexnum * 100))
c = int(idx / Vexnum * 100)
dur = time.perf_counter() - start
sys.stdout.write("\r{:^3.0f}%[{}=>{}]{:.2f}s".format(c, a, b, dur))
sys.stdout.flush()
with codecs.open(os.path.join(dir_path, file), 'r', encoding='utf-8') as load_f:
try:
text = json.load(load_f)
except:
falsefiles[file]=Vexname.index(file)-len(falsefiles)
continue
try:
links = []
for link in text['link']:
if link+'.json' in Vexname:
links.append(link)
text['link'] = links.copy()
except:
pass
finally:
if 'link' in text:
text['link'].clear()
else:
text['link'] = []
with codecs.open(os.path.join(dir_path, file), 'w', encoding='utf-8') as dump_f:
json.dump(text, dump_f, ensure_ascii=False,indent=4)
idx += 1
print('正在删除错误文件及链接...')
Vexnum -= len(falsefiles)
checknum=0
checkfalse=0
for file in list(falsefiles.keys()):
os.remove(os.path.join(dir_path,file))
Vexname.remove(file)
for i in range(checknum,falsefiles[file]):
with codecs.open(os.path.join(dir_path, Vexname[i]), 'r', encoding='utf-8') as load_f:
text = json.load(load_f)
try:
for falsefile in list(falsefiles.keys())[checkfalse:]:
if falsefile in text['link']:
text['link'].remove(falsefile)
except:
text['link'].clear()
with codecs.open(os.path.join(dir_path, Vexname[i]), 'w', encoding='utf-8') as dump_f:
json.dump(text, dump_f, ensure_ascii=False,indent=4)
checknum += falsefiles[file]
checkfalse += 1
# 计算PageRank
def Rank(Value, start):
NewValue=np.zeros(Vexnum,dtype=np.double)
for iter in range(1,epoch):
a = '=' * int(iter / epoch * 100)
b = ' ' * (100 - int(iter / epoch * 100))
c = int(iter / epoch * 100)
dur = time.perf_counter() - start
sys.stdout.write("\r{:^3.0f}%[{}=>{}]{:.2f}s".format(c, a, b, dur))
sys.stdout.flush()
for i in range(Vexnum):
with open(os.path.join(dir_path, Vexname[i]), 'r', encoding='utf-8') as load_f:
text = json.load(load_f)
count = len(text['link'])
if count == 0:
NewValue[i] = Value[i]
continue
for link in text['link']:
link += '.json'
NewValue[Vexname.index(link)] += Value[i] / count
for i in range(Vexnum):
NewValue[i] = NewValue[i] / (iter + 1) + Value[i] * (iter / (iter + 1))
Value=NewValue.copy()
return Value
def run():
print('开始计算PageRank...')
print('数据初始化...')
init()
Value = np.ones(len(Vexname),dtype=np.double)*(1000.0/Vexnum)
print('错误文件删除完毕！')
print('正在计算PageRank(迭代次数{})...'.format(epoch))
start = time.perf_counter()
Value = Rank(Value, start)
a = '=' * 100
b = ' ' * 0
c = 100
dur = time.perf_counter() - start
sys.stdout.write("\r{:^3.0f}%[{}=>{}]{:.2f}s".format(c, a, b, dur))
sys.stdout.flush()
print('\nPageRank计算完毕，正在往JSON中写入数据...')
max = {}
for file in Vexname: # 将PageRank写入JSON
with open(os.path.join(dir_path, file), 'r', encoding='utf-8') as load_f:
text = json.load(load_f)
with open(os.path.join(dir_path, file), 'w', encoding='utf-8') as dump_f:
text['weight'] = Value[Vexname.index(file)]
max[file] = text['weight']
json.dump(text, dump_f, ensure_ascii=False,indent=4)
print('数据写入完毕...')
if __name__ == '__main__':
run()# 导入必要的包
import json
import os
import sys
import time
import math
import gc
import elasticsearch
import numpy as np
import paddle.fluid as fluid
dir_path = os.path.dirname(os.path.realpath(__file__))
# 用训练好的模型进行预测并输出预测结果
# 创建执行器
place = fluid.CPUPlace()
exe = fluid.Executor(place)
exe.run(fluid.default_startup_program())
save_path = os.path.join(dir_path, 'infer_model/')
# 从模型中获取预测程序、输入数据名称列表、分类器
[infer_program, feeded_var_names, target_var] = fluid.io.load_inference_model(dirname=save_path, executor=exe)
# 主机
host = "py7hon.com:9200"
# 建立 elasticsearch 连接
try:
es = elasticsearch.Elasticsearch(hosts=host)
except Exception as e:
print(e)
exit()
# 获取数据
def get_data(sentence):
# 读取数据字典
with open(os.path.join(dir_path, 'dict_txt.txt'), 'r', encoding='utf-8') as f_data:
dict_txt = eval(f_data.readlines()[0])
dict_txt = dict(dict_txt)
# 把字符串数据转换成列表数据
keys = dict_txt.keys()
data = []
for s in sentence:
# 判断是否存在未知字符
if not s in keys:
s = '<unk>'
data.append((np.int64)(dict_txt[s]))
return data
def batch_reader(Json_list,json_path):
datas = []
gc.collect()
json_files = []
falsefiles = []
datas.clear()
falsefiles.clear()
json_files.clear()
start = time.perf_counter()
i=0
scale = 100
for file in Json_list:
if i % 100 == 0:
a = '=' * int(i / len(Json_list) * 100)
b = ' ' * (scale - int(i / len(Json_list) * 100))
c = int(i / len(Json_list) * 100)
dur = time.perf_counter() - start
sys.stdout.write("\r{:^3.0f}%[{}=>{}]{:.2f}s".format(c, a, b, dur))
sys.stdout.flush()
i+=1
with open(os.path.join(json_path, file), "r", encoding='utf-8') as f:
try:
text = json.load(f)
except:
falsefiles.append(file)
continue
json_files.append(os.path.join(json_path, file))
json_text = text['content']
data = get_data(json_text)
datas.append(data)
for file in falsefiles:
os.remove(os.path.join(dir_path, file))
file_count = len(Json_list) - len(falsefiles)
a = '=' * 100
b = ' ' * 0
c = 100
dur = time.perf_counter() - start
sys.stdout.write("\r{:^3.0f}%[{}=>{}]{:.2f}s".format(c, a, b, dur))
sys.stdout.flush()
print('\n文本数据获取完毕，共计{0}条文本数据，有效数据{2}条，无效数据{1}条（已删除）！'.format(len(Json_list),len(falsefiles),file_count))
print('AI正在加载分类模型...')
# 获取每句话的单词数量
base_shape = [[len(c) for c in datas]]
# 生成预测数据
tensor_words = fluid.create_lod_tensor(datas, base_shape, place)
# 执行预测
result = exe.run(program=infer_program,
feed={feeded_var_names[0]: tensor_words},
fetch_list=target_var)
print('模型加载完毕！')
# 分类名称
names = ['文化', '娱乐', '体育', '财经', '房产', '汽车', '教育', '科技', '国际', '证券']
count = np.zeros(10)
print('AI正在对文本数据进行分类并上传至ES：')
# 获取结果概率最大的label
start = time.perf_counter()
for i in range(file_count):
if i % 100 == 0:
a = '=' * int(i / file_count * 100)
b = ' ' * (scale - int(i / file_count * 100))
c = int(i / file_count * 100)
dur = time.perf_counter() - start
sys.stdout.write("\r{:^3.0f}%[{}=>{}]{:.2f}s".format(c, a, b, dur))
sys.stdout.flush()
lab = np.argsort(result)[0][i][-1]
# print('预测结果标签为：%d， 名称为：%s， 概率为：%f' % (lab, names[lab], result[0][i][lab]))
count[lab] += 1
with open(json_files[i], 'r', encoding='utf-8') as load_f:
try:
text = json.load(load_f)
except:
continue
text['content_type'] = names[lab]
id = json_files[i].split('\\')[-1].split('.')[0]
#try:
del text['link']
response = es.index(index='page', doc_type='_doc', id=id, body=text)
#except Exception:
# print("\n" + "数据 " + id + " 插入失败，错误信息：" + response)
# with open(os.path.join(json_path,json_files[i].split('\\')[-1]),'w') as dump_f:
# json.dump(text,dump_f)
a = '=' * 100
b = ' ' * 0
c = 100
dur = time.perf_counter() - start
sys.stdout.write("\r{:^3.0f}%[{}=>{}]{:.2f}s".format(c, a, b, dur))
sys.stdout.flush()
print("\n" + "%d条文本数据分类结束！已全部上传至ES" % (file_count))
def run():
# 获取图片数据
print('AI正在获取文本数据...')
json_path = os.path.realpath(__file__) + '/../../../RawData'
Json_list = os.listdir(json_path)
batch_size=500
if len(Json_list)>batch_size:
Json_batch=0
print('当前文本数量为{0}条，正在分批处理...'.format(len(Json_list)))
for batch_id in range(math.ceil(len(Json_list)/batch_size)):
a=(batch_size if batch_size<(len(Json_list)-Json_batch) else len(Json_list)-Json_batch)
print('正在处理第{0}批，数量为{1}...'.format(batch_id+1,a))
batch_reader(Json_list[Json_batch:Json_batch+a],json_path)
Json_batch += a
else:
batch_reader(Json_list,json_path)
if __name__ == '__main__':
run()4.系统展示
