flask快速开发框架(手把手教你使用Flask搭建ES搜索引擎预备篇)
flask快速开发框架
手把手教你使用Flask搭建ES搜索引擎预备篇1 前言
Elasticsearch 是一个开源的搜索引擎,建立在一个全文搜索引擎库 Apache Lucene™ 基础之上。
那么如何实现 Elasticsearch和 Python 的对接成为我们所关心的问题了 (怎么什么都要和 Python 关联啊)。
2 Python 交互所以,Python 也就提供了可以对接 Elasticsearch的依赖库。
- pip install elasticsearch
初始化连接一个 Elasticsearch 操作对象。
- def __init__(self, index_type: str, index_name: str, ip="127.0.0.1"):
- # self.es = Elasticsearch([ip], http_auth=('username', 'password'), port=9200)
- self.es = Elasticsearch("localhost:9200")
- self.index_type = index_type
- self.index_name = index_name
默认端口 9200,初始化前请确保本地已搭建好 Elasticsearch的所属环境。
根据 ID 获取文档数据- def get_doc(self, uid):
- return self.es.get(index=self.index_name, id=uid)
- def insert_one(self, doc: dict):
- self.es.index(index=self.index_name, doc_type=self.index_type, body=doc)
- def insert_array(self, docs: list):
- for doc in docs:
- self.es.index(index=self.index_name, doc_type=self.index_type, body=doc)
- def search(self, query, count: int = 30):
- dsl = {
- "query": {
- "multi_match": {
- "query": query,
- "fields": ["title", "content", "link"]
- }
- },
- "highlight": {
- "fields": {
- "title": {}
- }
- }
- }
- match_data = self.es.search(index=self.index_name, body=dsl, size=count)
- return match_data
- def __search(self, query: dict, count: int = 20): # count: 返回的数据大小
- results = []
- params = {
- 'size': count
- }
- match_data = self.es.search(index=self.index_name, body=query, params=params)
- for hit in match_data['hits']['hits']:
- results.append(hit['_source'])
- return results
- def delete_index(self):
- try:
- self.es.indices.delete(index=self.index_name)
- except:
- pass
好啊,封装 search 类也是为了方便调用,整体贴一下。
- from elasticsearch import Elasticsearch
- class elasticSearch():
- def __init__(self, index_type: str, index_name: str, ip="127.0.0.1"):
- # self.es = Elasticsearch([ip], http_auth=('elastic', 'password'), port=9200)
- self.es = Elasticsearch("localhost:9200")
- self.index_type = index_type
- self.index_name = index_name
- def create_index(self):
- if self.es.indices.exists(index=self.index_name) is True:
- self.es.indices.delete(index=self.index_name)
- self.es.indices.create(index=self.index_name, ignore=400)
- def delete_index(self):
- try:
- self.es.indices.delete(index=self.index_name)
- except:
- pass
- def get_doc(self, uid):
- return self.es.get(index=self.index_name, id=uid)
- def insert_one(self, doc: dict):
- self.es.index(index=self.index_name, doc_type=self.index_type, body=doc)
- def insert_array(self, docs: list):
- for doc in docs:
- self.es.index(index=self.index_name, doc_type=self.index_type, body=doc)
- def search(self, query, count: int = 30):
- dsl = {
- "query": {
- "multi_match": {
- "query": query,
- "fields": ["title", "content", "link"]
- }
- },
- "highlight": {
- "fields": {
- "title": {}
- }
- }
- }
- match_data = self.es.search(index=self.index_name, body=dsl, size=count)
- return match_data
尝试一下把 Mongodb 中的数据插入到 ES 中。
- import json
- from datetime import datetime
- import pymongo
- from app.elasticsearchClass import elasticSearch
- client = pymongo.MongoClient('127.0.0.1', 27017)
- db = client['spider']
- sheet = db.get_collection('Spider').find({}, {'_id': 0, })
- es = elasticSearch(index_type="spider_data",index_name="spider")
- es.create_index()
- for i in sheet:
- data = {
- 'title': i["title"],
- 'content':i["data"],
- 'link': i["link"],
- 'create_time':datetime.now()
- }
- es.insert_one(doc=data)
到 ES 中查看一下,启动 elasticsearch-head 插件。
如果是 npm 安装的那么 cd 到根目录之后直接 npm run start 就跑起来了。
本地访问 http://localhost:9100/
发现新加的 spider 数据文档确实已经进去了。
3 爬虫入库要想实现 ES 搜索,首先要有数据支持,而海量的数据往往来自爬虫。
为了节省时间,编写一个最简单的爬虫,抓取 百度百科。
简单粗暴一点,先 递归获取 很多很多的 url 链接
- import requests
- import re
- import time
- exist_urls = []
- headers = {
- 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.62 Safari/537.36',
- }
- def get_link(url):
- try:
- response = requests.get(url=url, headers=headers)
- response.encoding = 'UTF-8'
- html = response.text
- link_lists = re.findall('.*?<a target=_blank href="/item/([^:#=<>]*?)".*?</a>', html)
- return link_lists
- except Exception as e:
- pass
- finally:
- exist_urls.append(url)
- # 当爬取深度小于10层时,递归调用主函数,继续爬取第二层的所有链接
- def main(start_url, depth=1):
- link_lists = get_link(start_url)
- if link_lists:
- unique_lists = list(set(link_lists) - set(exist_urls))
- for unique_url in unique_lists:
- unique_url = 'https://baike.baidu.com/item/' + unique_url
- with open('url.txt', 'a+') as f:
- f.write(unique_url + '\n')
- f.close()
- if depth < 10:
- main(unique_url, depth + 1)
- if __name__ == '__main__':
- start_url = 'https://baike.baidu.com/item/%E7%99%BE%E5%BA%A6%E7%99%BE%E7%A7%91'
- main(start_url)
把全部 url 存到 url.txt 文件中之后,然后启动任务。
- # parse.py
- from celery import Celery
- import requests
- from lxml import etree
- import pymongo
- app = Celery('tasks', broker='redis://localhost:6379/2')
- client = pymongo.MongoClient('localhost',27017)
- db = client['baike']
- @app.task
- def get_url(link):
- item = {}
- 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'}
- res = requests.get(link,headers=headers)
- res.encoding = 'UTF-8'
- doc = etree.HTML(res.text)
- content = doc.xpath("//li[@class='lemma-summary']/li[@class='para']//text()")
- print(res.status_code)
- print(link,'\t','++++++++++++++++++++')
- item['link'] = link
- data = ''.join(content).replace(' ', '').replace('\t', '').replace('\n', '').replace('\r', '')
- item['data'] = data
- if db['Baike'].insert(dict(item)):
- print("is OK ...")
- else:
- print('Fail')
- from parse import get_url
- def main(url):
- result = get_url.delay(url)
- return result
- def run():
- with open('./url.txt', 'r') as f:
- for url in f.readlines():
- main(url.strip('\n'))
- if __name__ == '__main__':
- run()
- 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
- python线程池的实现原理(详解python中的线程与线程池)
- python 后台django(Python Django给admin添加Action的方法实例详解)
- python time模块记录时间(Python Datetime模块和Calendar模块用法实例分析)
- python数组矩阵操作(Python矩阵和Numpy数组的那些事儿)
- eval在python里面什么意思(python 为什么说eval要慎用)
- python飞机大战游戏背景(python实现飞机大战游戏)
- python淘宝秒杀教程(Python实现京东秒杀功能代码)
- python微信防封(深入学习微信网址链接解封的防封原理visit_type)
- python设计一个聊天机器人(手把手教你使用Python创建微信机器人)
- python中递归方法(Python中最大递归深度值的探讨)
- python中if的条件语句(浅谈Python的条件判断语句if/else语句)
- python多线程和多进程应用场景(Python多线程处理实例详解单进程/多进程)
- python读取mat文件(详解Python Matplot中文显示完美解决方案)
- python开启多线程(python 多线程重启方法)
- python读取txt文件每一行数据(python读取txt文件并取其某一列数据的示例)
- python中jieba库怎么用(详解Python数据可视化编程 - 词云生成并保存jieba+WordCloud)
- 痴情男神 吴彦祖 与妻子恋爱8年,结婚10年,家庭幸福美满(痴情男神吴彦祖)
- 成功破圈,小牛电动SQi强势开 跨(小牛电动SQi强势开)
- 挑战新国标电自天花板,九号机械师MMAX 110P深度体验(挑战新国标电自天花板)
- 《满江红》不要只当电影看,学生应该这样做(满江红不要只当电影看)
- 电影《民间怪谈录之走阴人》定档8月5日,开启一场中式惊悚之旅(电影民间怪谈录之走阴人定档8月5日)
- 原创图画书,以儿童视角讲述中国故事(以儿童视角讲述中国故事)
热门推荐
- css修饰文本的方法(CSS常用文本样式总结)
- dede常用变量(dede调用文章第一张图片非缩略图的实现方法)
- php检查语法(使用 PHP Masked Package 屏蔽敏感数据的实现方法)
- lombok 代码行数(Lombok实现方式JSR-269)
- 属于web服务器的有哪些(web服务器有几种类型?)
- python3正则表达式详解(Python正则表达式和re库知识点总结)
- mysqlupdate怎么设置(MySQL update 语句的正确用法)
- docker配置和管理(Docker 容器监控原理及 cAdvisor的安装与使用说明)
- mysql怎么建立主键外键(mysql 实现设置多个主键的操作)
- dedecms的dedesql.class.php on line 489错误的解决方法(dedecms的dedesql.class.php on line 489错误的解决方法)
排行榜
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9