python操作elasticseach

青旅半醒 2022-03-09 04:27 179阅读 0赞

#!/usr/bin/env python
    # encoding: utf-8
    """
    @author: wanwei
    @license: (C) Copyright 2013-2017, Node Supply Chain Manager Corporation Limited.
    @contact: 
    @software: pycharm
    @file: elasticsearch.py
    @time: 2018/10/10 10:38
    @desc:
    """
    
    import os
    import time
    from os import walk
    import csv
    import json
    from datetime import datetime
    from elasticsearch import Elasticsearch
    from elasticsearch.helpers import bulk
    
    
    class ElasticObj:
        def __init__(self, index_name, index_type, ip="127.0.0.1"):
            '''
    
            :param index_name: 索引名称
            :param index_type: 索引类型
            '''
            self.index_name = index_name
            self.index_type = index_type
            # 无用户名密码状态
            #self.es = Elasticsearch([ip])
            #用户名密码状态
            self.es = Elasticsearch([ip],port='9201')
    
        def create_index(self):
            '''
            创建索引,创建索引名称为ott，类型为ott_type的索引
            :param ex: Elasticsearch对象
            :return:
            '''
            #创建映射
            _index_mappings = {
                "mappings": {
                    self.index_type: {
                        "properties": {
                            "title": {
                                "type": "text",
                                "index": True,
                                "analyzer": "ik_max_word",
                                "search_analyzer": "ik_max_word"
                            },
                            "date": {
                                "type": "text",
                                "index": True
                            },
                            "keyword": {
                                "type": "string",
                                "index": "not_analyzed"
                            },
                            "source": {
                                "type": "string",
                                "index": "not_analyzed"
                            },
                            "link": {
                                "type": "string",
                                "index": "not_analyzed"
                            }
                        }
                    }
    
                }
            }
            if self.es.indices.exists(index=self.index_name) is not True:
                res = self.es.indices.create(index=self.index_name, body=_index_mappings)
                print (res)
    
    
        def IndexData(self):
            es = Elasticsearch()
            csvdir = 'D:/work/ElasticSearch/exportExcels'
            filenamelist = []
            for (dirpath, dirnames, filenames) in walk(csvdir):
                filenamelist.extend(filenames)
                break
            total = 0
            for file in filenamelist:
                csvfile = csvdir + '/' + file
                self.Index_Data_FromCSV(csvfile,es)
                total += 1
                print (total)
                time.sleep(10)
    
        def Index_Data_FromCSV(self,csvfile):
            '''
            从CSV文件中读取数据，并存储到es中
            :param csvfile: csv文件，包括完整路径
            :return:
            '''
            with open(csvfile) as f:
                data_list = csv.DictReader(f)
                for item in data_list:
                    res = self.es.index(index=self.index_name, doc_type=self.index_type, body=item)
                    print(res['result'])
    
        def Index_Data(self):
            '''
            数据存储到es
            :return:
            '''
            list = [{'id_card_no': '44010119900****123', 'disc_person': '0', 'zb': '1', 'id': '08e7589175745d85569c3a97b3c6dae2', 'name': '2', 'update_time': '2018-10-10 15:20:55', 'disc_consume': '1', 'disc_trans': '1', 'execute_person': '0'}]
            for item in list:
                res = self.es.index(index=self.index_name, doc_type=self.index_type, body=item)
                # print(res['created'])
                print(res)
    
        def bulk_Index_Data(self):
            '''
            用bulk将批量数据存储到es
            :return:
            '''
            list = [
                {"date": "2017-09-13",
                 "source": "慧聪网",
                 "link": "http://info.broadcast.hc360.com/2017/09/130859749974.shtml",
                 "keyword": "电视",
                 "title": "付费 电视 行业面临的转型和挑战"
                 },
                {"date": "2017-09-13",
                 "source": "中国文明网",
                 "link": "http://www.wenming.cn/xj_pd/yw/201709/t20170913_4421323.shtml",
                 "keyword": "电视",
                 "title": "电视 专题片《巡视利剑》广获好评：铁腕反腐凝聚党心民心"
                 },
                {"date": "2017-09-13",
                 "source": "人民电视",
                 "link": "http://tv.people.com.cn/BIG5/n1/2017/0913/c67816-29533981.html",
                 "keyword": "电视",
                 "title": "中国第21批赴刚果（金）维和部隊启程--人民 电视 --人民网"
                 },
                {"date": "2017-09-13",
                 "source": "站长之家",
                 "link": "http://www.chinaz.com/news/2017/0913/804263.shtml",
                 "keyword": "电视",
                 "title": "电视 盒子 哪个牌子好？ 吐血奉献三大选购秘笈"
                 }
            ]
            ACTIONS = []
            i = 1
            for line in list:
                action = {
                    "_index": self.index_name,
                    "_type": self.index_type,
                    "_id": i, #_id 也可以默认生成，不赋值
                    "_source": {
                        "date": line['date'],
                        "source": line['source'].decode('utf8'),
                        "link": line['link'],
                        "keyword": line['keyword'].decode('utf8'),
                        "title": line['title'].decode('utf8')}
                }
                i += 1
                ACTIONS.append(action)
                # 批量处理
            success, _ = bulk(self.es, ACTIONS, index=self.index_name, raise_on_error=True)
            print('Performed %d actions' % success)
    
        def Delete_Index_Data_By_Id(self,id):
            '''
            删除索引中的一条
            :param id:
            :return:
            '''
            res = self.es.delete(index=self.index_name, doc_type=self.index_type, id=id)
            print (res)
    
        def Delete_Index(self):
            '''
            删除index
            '''
            res = self.es.indices.delete(index=self.index_name, ignore=[400, 404])
            print (json.dumps(res, indent=2))
    
        def Get_Data_By_Id(self,id):
    
            res = self.es.get(index=self.index_name, doc_type=self.index_type,id=id)
            print(res['_source'])
    
            print ('------------------------------------------------------------------')
            #
            # # 输出查询到的结果
            # for hit in res['hits']['hits']:
            #     # print hit['_source']
            #     print (hit['_source']['date'],hit['_source']['source'],hit['_source']['link'],hit['_source']['keyword'],hit['_source']['title'])
    
    def get_data_by_body(index, size=10, must=[], must_not=[], should=[], page=1, sort=[], aggs={}):
        """
        使用JSON风格查询语言执行查询，bool查询允许我们使用布尔逻辑将较小的查询组合成较大的查询,
        must相当于and，should相当于or，must_not相当于!
        :param index: 索引
        :param must: 相当于逻辑and
        返回所有文档："must":  [{"match_all": {}}]
        模糊匹配age为40的文档："must": [{ "match": { "age": "40" } }]
        :param must_not: 相当于逻辑!
                "must_not":[
                    { "match": { "state": "ID" } }
                  ]
        :param should: 相当于or
                 "should": [
                        { "match": { "address": "mill" } },
                        { "match": { "address": "lane" } }
                      ]
        :param begin: 参数（从0开始）指定从哪个文档索引开始
        :param size: 指定从begin开始返回多少条
        :param sort: 对返回结果进行排序，如下：按balance对结果进行降序排序
                "sort":{
              "query": { "match_all": {} },
              "sort": { "balance": { "order": "desc" } }
                 }
        :param aggs: 按state对所有帐户进行分组，然后按照count数降序（默认）返回前10条（默认）：
                  "aggs": {
                        "group_by_state": {
                          "terms": {
                            "field": "state.keyword"
                          }
                        }
                      }
        :return: 返回命中结果集列表
        """
        if not current_app.elasticsearch:
            return
        doc = {"query": {"bool": {"must": must, "must_not": must_not, "should": should}}, "from": (page-1)*size, "size": size,
               "sort": sort, "aggs": aggs}
        _searched = current_app.elasticsearch.search(index=index, body=doc)
        results = []
        for hit in _searched['hits']['hits']:
            results.append(hit.get('_source', ''))
        return results
    
    
    obj = ElasticObj("test","test",ip ="x.x.x.x")
    #删除索引
    obj.Delete_Index()
    #按查询参数查询
    # obj.Get_Data_By_Body()
    #按id查询
    # obj.Get_Data_By_Id("be8212c27f091d8c63d33efb1686ba0f")
    #创建索引
    # obj.create_index()
    # obj.Index_Data()
    # obj.bulk_Index_Data()
    # obj.IndexData()
    # obj.Delete_Index_Data_By_Id(1)
    # csvfile = 'D:/work/ElasticSearch/exportExcels/2017-08-31_info.csv'
    # obj.Index_Data_FromCSV(csvfile)
    # obj.GetData(es)