手把手教你使用Flask搭建ES搜索引擎(预备篇)

1 前言

Elasticsearch 是一个开源的搜索引擎，建立在一个全文搜索引擎库 Apache Lucene™ 基础之上。

那么如何实现 Elasticsearch和 Python 的对接成为我们所关心的问题了 (怎么什么都要和 Python 关联啊)。

2 Python 交互

所以，Python 也就提供了可以对接 Elasticsearch的依赖库。

1. pip install elasticsearch 

初始化连接一个 Elasticsearch 操作对象。

1. def __init__(self, index_type: str, index_name: str, ip="127.0.0.1"): 
2.  
3.     # self.es = Elasticsearch([ip], http_auth=('username', 'password'), port=9200) 
4.     self.es = Elasticsearch("localhost:9200") 
5.     self.index_type = index_type 
6.     self.index_name = index_name 

默认端口 9200，初始化前请确保本地已搭建好 Elasticsearch的所属环境。

根据 ID 获取文档数据

1. def get_doc(self, uid): 
2.     return self.es.get(index=self.index_name, id=uid) 

插入文档数据

1. def insert_one(self, doc: dict): 
2.     self.es.index(index=self.index_name, doc_type=self.index_type, body=doc) 
3.  
4. def insert_array(self, docs: list): 
5.     for doc in docs: 
6.         self.es.index(index=self.index_name, doc_type=self.index_type, body=doc) 

搜索文档数据

1. def search(self, query, count: int = 30): 
2.     dsl = { 
3.         "query": { 
4.             "multi_match": { 
5.                 "query": query, 
6.                 "fields": ["title", "content", "link"] 
7.             } 
8.         }, 
9.         "highlight": { 
10.             "fields": { 
11.                 "title": {} 
12.             } 
13.         } 
14.     } 
15.     match_data = self.es.search(index=self.index_name, body=dsl, size=count) 
16.     return match_data 
17.  
18. def __search(self, query: dict, count: int = 20): # count: 返回的数据大小 
19.     results = [] 
20.     params = { 
21.         'size': count 
22.     } 
23.     match_data = self.es.search(index=self.index_name, body=query, params=params) 
24.     for hit in match_data['hits']['hits']: 
25.         results.append(hit['_source']) 
26.  
27.     return results 

删除文档数据

1. def delete_index(self): 
2.     try: 
3.         self.es.indices.delete(index=self.index_name) 
4.     except: 
5.         pass 

好啊，封装 search 类也是为了方便调用，整体贴一下。

1. from elasticsearch import Elasticsearch 
2.  
3.  
4. class elasticSearch(): 
5.  
6.     def __init__(self, index_type: str, index_name: str, ip="127.0.0.1"): 
7.  
8.         # self.es = Elasticsearch([ip], http_auth=('elastic', 'password'), port=9200) 
9.         self.es = Elasticsearch("localhost:9200") 
10.         self.index_type = index_type 
11.         self.index_name = index_name 
12.  
13.     def create_index(self): 
14.         if self.es.indices.exists(index=self.index_name) is True: 
15.             self.es.indices.delete(index=self.index_name) 
16.         self.es.indices.create(index=self.index_name, ignore=400) 
17.  
18.     def delete_index(self): 
19.         try: 
20.             self.es.indices.delete(index=self.index_name) 
21.         except: 
22.             pass 
23.  
24.     def get_doc(self, uid): 
25.         return self.es.get(index=self.index_name, id=uid) 
26.  
27.     def insert_one(self, doc: dict): 
28.         self.es.index(index=self.index_name, doc_type=self.index_type, body=doc) 
29.  
30.     def insert_array(self, docs: list): 
31.         for doc in docs: 
32.             self.es.index(index=self.index_name, doc_type=self.index_type, body=doc) 
33.  
34.     def search(self, query, count: int = 30): 
35.         dsl = { 
36.             "query": { 
37.                 "multi_match": { 
38.                     "query": query, 
39.                     "fields": ["title", "content", "link"] 
40.                 } 
41.             }, 
42.             "highlight": { 
43.                 "fields": { 
44.                     "title": {} 
45.                 } 
46.             } 
47.         } 
48.         match_data = self.es.search(index=self.index_name, body=dsl, size=count) 
49.         return match_data 

尝试一下把 Mongodb 中的数据插入到 ES 中。

1. import json 
2. from datetime import datetime 
3. import pymongo 
4. from app.elasticsearchClass import elasticSearch 
5.  
6. client = pymongo.MongoClient('127.0.0.1', 27017) 
7. db = client['spider'] 
8. sheet = db.get_collection('Spider').find({}, {'_id': 0, }) 
9.  
10. es = elasticSearch(index_type="spider_data",index_name="spider") 
11. es.create_index() 
12.  
13. for i in sheet: 
14.     data = { 
15.             'title': i["title"], 
16.             'content':i["data"], 
17.             'link': i["link"], 
18.             'create_time':datetime.now() 
19.         } 
20.  
21.     es.insert_one(doc=data) 

到 ES 中查看一下，启动 elasticsearch-head 插件。

如果是 npm 安装的那么 cd 到根目录之后直接 npm run start 就跑起来了。

本地访问 http://localhost:9100/

发现新加的 spider 数据文档确实已经进去了。

3 爬虫入库

要想实现 ES 搜索，首先要有数据支持，而海量的数据往往来自爬虫。

为了节省时间，编写一个最简单的爬虫，抓取 百度百科。

简单粗暴一点，先 递归获取 很多很多的 url 链接

1. import requests 
2. import re 
3. import time 
4.  
5. exist_urls = [] 
6. headers = { 
7.     'User-Agent': 'Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.62 Safari/537.36', 
8. } 
9.  
10. def get_link(url): 
11.     try: 
12.         response = requests.get(url=url, headers=headers) 
13.         response.encoding = 'UTF-8' 
14.         html = response.text 
15.         link_lists = re.findall('.*?<a target=_blank href="/item/([^:#=<>]*?)".*?</a>', html) 
16.         return link_lists 
17.     except Exception as e: 
18.         pass 
19.     finally: 
20.         exist_urls.append(url) 
21.  
22.  
23. # 当爬取深度小于10层时，递归调用主函数，继续爬取第二层的所有链接 
24. def main(start_url, depth=1): 
25.     link_lists = get_link(start_url) 
26.     if link_lists: 
27.         unique_lists = list(set(link_lists) - set(exist_urls)) 
28.         for unique_url in unique_lists: 
29.             unique_url = 'https://baike.baidu.com/item/' + unique_url 
30.  
31.             with open('url.txt', 'a+') as f: 
32.                 f.write(unique_url + '\n') 
33.                 f.close() 
34.         if depth < 10: 
35.             main(unique_url, depth + 1) 
36.  
37. if __name__ == '__main__': 
38.     start_url = 'https://baike.baidu.com/item/%E7%99%BE%E5%BA%A6%E7%99%BE%E7%A7%91' 
39.     main(start_url) 

把全部 url 存到 url.txt 文件中之后，然后启动任务。

1. # parse.py 
2. from celery import Celery 
3. import requests 
4. from lxml import etree 
5. import pymongo 
6. app = Celery('tasks', broker='redis://localhost:6379/2') 
7. client = pymongo.MongoClient('localhost',27017) 
8. db = client['baike'] 
9. @app.task 
10. def get_url(link): 
11.     item = {} 
12.     headers = {'User-Agent':'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/34.0.1847.131 Safari/537.36'} 
13.     res = requests.get(link,headers=headers) 
14.     res.encoding = 'UTF-8' 
15.     doc = etree.HTML(res.text) 
16.     content = doc.xpath("//div[@class='lemma-summary']/div[@class='para']//text()") 
17.     print(res.status_code) 
18.     print(link,'\t','++++++++++++++++++++') 
19.     item['link'] = link 
20.     data = ''.join(content).replace(' ', '').replace('\t', '').replace('\n', '').replace('\r', '') 
21.     item['data'] = data 
22.     if db['Baike'].insert(dict(item)): 
23.         print("is OK ...") 
24.     else: 
25.         print('Fail') 

run.py 飞起来

1. from parse import get_url 
2.  
3. def main(url): 
4.     result = get_url.delay(url) 
5.     return result 
6.  
7. def run(): 
8.     with open('./url.txt', 'r') as f: 
9.         for url in f.readlines(): 
10.             main(url.strip('\n')) 
11.  
12. if __name__ == '__main__': 
13.     run() 

黑窗口键入

1. celery -A parse worker -l info -P gevent -c 10 

哦豁 !! 你居然使用了 Celery 任务队列，gevent 模式，-c 就是10个线程刷刷刷就干起来了，速度杠杠的 !!

啥?分布式? 那就加多几台机器啦，直接把代码拷贝到目标服务器，通过 redis 共享队列协同多机抓取。

这里是先将数据存储到了 MongoDB 上(个人习惯)，你也可以直接存到 ES 中，但是单条单条的插入速度堪忧(接下来会讲到优化，哈哈)。

使用前面的例子将 Mongo 中的数据批量导入到 ES 中，OK !!!

到这一个简单的数据抓取就已经完毕了。

好啦，现在 ES 中已经有了数据啦，接下来就应该是 Flask web 的操作啦，当然，Django，FastAPI 也很优秀。嘿嘿，你喜欢 !!

原文链接：https://mp.weixin.qq.com/s/kZBcS-9esICzfS7C_tmovA