package com.zzsn.common.es;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.MainResponse;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author kongliufeng
 * @create 2020-08-21 12:34
 */
@Slf4j
@Component
public class ElasticSearchClient {

    @Resource
    ElasticsearchClientConfig elasticsearchClientConfig;

    @Resource(name = "restHighLevelClientEs")
    RestHighLevelClient client;

    public String getIndex(){
        return elasticsearchClientConfig.getIndexs();
    }
    /**
     * 认证DEMO服务的SERVICEID
     */
    public static final String textExtract = "textExtract";
    /**
     * 获取节点相关信息
     * @return
     */
    public Map<String ,Object> getEsInfo(){
        try {
            Map<String ,Object> map =new HashMap<>();
            //获取Es相关集群信息
            MainResponse response = client.info(RequestOptions.DEFAULT);
            String clusterName = response.getClusterName();
            String clusterUuid = response.getClusterUuid();
            String nodeName = response.getNodeName();
            MainResponse.Version version = response.getVersion();
            String buildDate = version.getBuildDate();
            String buildFlavor = version.getBuildFlavor();
            String buildHash = version.getBuildHash();
            String buildType = version.getBuildType();
            String luceneVersion = version.getLuceneVersion();
            String minimumIndexCompatibilityVersion= version.getMinimumIndexCompatibilityVersion();
            String minimumWireCompatibilityVersion = version.getMinimumWireCompatibilityVersion();
            String number = version.getNumber();
            map.put("clusterName",clusterName);
            map.put("clusterUuid",clusterUuid);
            map.put("nodeName",nodeName);
            map.put("version",version);
            map.put("buildDate",buildDate);
            map.put("buildFlavor",buildFlavor);
            map.put("buildHash",buildHash);
            map.put("buildType",buildType);
            map.put("luceneVersion",luceneVersion);
            map.put("minimumIndexCompatibilityVersion",minimumIndexCompatibilityVersion);
            map.put("minimumWireCompatibilityVersion",minimumWireCompatibilityVersion);
            map.put("number",number);
            return map;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 获取低级客户端
     * @return
     */
    public RestClient getLowLevelClient(){
        return client.getLowLevelClient();
    }

    /**
     * 创建索引
     * @param index
     * @return
     */
    public boolean indexCreate(String index){
        try {
            if(!indexExist(index)){
                CreateIndexRequest request = new CreateIndexRequest(index);
                CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
                log.info(createIndexResponse.isAcknowledged()?"创建索引[{}]成功":"创建索引[{}]失败",index);
                return createIndexResponse.isAcknowledged();
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 判断索引是否存在
     * @param indices
     * @return
     */
    @SneakyThrows
    public boolean indexExist(String... indices){
        GetIndexRequest request = new GetIndexRequest(indices);
        return  client.indices().exists(request,RequestOptions.DEFAULT);
    }

    /**
     * 删除索引
     * @param index
     * @return
     */
    @SneakyThrows
    public boolean deleteIndex(String index){
        DeleteIndexRequest request = new DeleteIndexRequest(index);
        AcknowledgedResponse delete = client.indices().delete(request,RequestOptions.DEFAULT);
        return delete.isAcknowledged();
    }

    public void docGet(String index,String id) {
        try {
            GetRequest getRequest = new GetRequest(index, id);
            GetResponse documentFields = client.get(getRequest, RequestOptions.DEFAULT);
            Map<String, Object> sourceAsMap = documentFields.getSourceAsMap();
            System.out.println(documentFields.toString());
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    /**
     * 判断doc是否存在
     * @param index
     * @param id
     * @return
     */
    @SneakyThrows
    public boolean docExists(String index,String id){
        GetRequest request = new GetRequest(index,id);
        //禁用提取_source
        request.fetchSourceContext(new FetchSourceContext(false));
        //禁用获取存储的字段
        request.storedFields("_none_");
        boolean exists = client.exists(request, RequestOptions.DEFAULT);
        return  exists;
    }

    /**
     *  批量插入数据
     * @param index 索引库
     * @param list  List<ESBaseData> 批量保存的list,根据实际需求实体集成ESBaseData
     */
    public void docSavaBulk(String index, List<ESBaseData> list){
        BulkRequest request = new BulkRequest();
        request.timeout(TimeValue.timeValueMinutes(10));
        for (int i = 0; i < list.size(); i++) {
            ESBaseData b = list.get(i);
            request.add(new IndexRequest(index).id(b.getId()).source(
                    JSON.toJSONString(list.get(i)), XContentType.JSON
            ));
        }
        try {
            BulkResponse bulk = client.bulk(request, RequestOptions.DEFAULT);
            BulkItemResponse[] bulkItemResponses = bulk.getItems();
            int length = bulkItemResponses.length;
            for(int i = 0; i < length; ++i) {
                BulkItemResponse response = bulkItemResponses[i];
                if (response.isFailed()) {//查看所有请求失败结果
                    log.info("批量保存[{}]过程中,id为[{}]的保存失败,失败原因[{}]",response.getIndex(),response.getId(),response.getFailureMessage());
                }else{//请求成功的
                    log.info("批量保存[{}]过程中,id为[{}]的保存成功,状态[{}],version[{}]",response.getIndex(),response.getId(),response.status(),response.getVersion());
                }
            }
        }catch (IOException e){
            e.printStackTrace();
            log.warn("批量[{}]保存失败",index);
        }
    }

    /**
     *  批量插入数据(异步)
     * @param index 索引库
     * @param list  List<ESBaseData> 批量保存的list,根据实际需求实体集成ESBaseData
     */
    public void docSavaBulkAsync(String index,List<ESBaseData> list){
        BulkRequest request = new BulkRequest();
        request.timeout(TimeValue.timeValueMinutes(10));
        for (int i = 0; i < list.size(); i++) {
            ESBaseData b = list.get(i);
            request.add(new IndexRequest(index).id(b.getId()).source(
                    JSON.toJSONString(list.get(i)),XContentType.JSON
            ));
        }
        client.bulkAsync(request, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse bulkItemResponses) {
                BulkItemResponse[] bulkItems = bulkItemResponses.getItems();
                int length = bulkItems.length;
                for(int i = 0; i < length; ++i) {
                    BulkItemResponse response = bulkItems[i];
                    if (response.isFailed()) {//查看所有请求失败结果
                        log.info("批量保存[{}]过程中,id为[{}]的保存失败,失败原因[{}]",response.getIndex(),response.getId(),response.getFailureMessage());
                    }else{//请求成功的
                        log.info("批量保存[{}]过程中,id为[{}]的保存成功,状态[{}],version[{}]",response.getIndex(),response.getId(),response.status(),response.getVersion());
                    }
                }
            }
            @Override
            public void onFailure(Exception e) {
                log.warn("批量[{}]保存失败,失败原因[{}]",index,e.getMessage());
            }
        });
    }

    /**
     * 插入数据
     * @param index
     * @param id
     * @param object
     * @return
     */
    public String docSavaByEntity(String index,String id,Object object){
        //System.out.println(JSON.toJSONString(object));
        return docSaveByJson(index,id,JSON.toJSONString(object, SerializerFeature.WriteMapNullValue));
    }

    public void docSavaByEntityAsync(String index,String id,Object object){
        docSaveByJsonAsync(index,id,JSON.toJSONString(object,SerializerFeature.WriteMapNullValue));
    }

    /**
     * 保存json
     * @param index
     * @param id
     * @param jsonStr
     * @return
     */
    public String docSaveByJson(String index, String id, String jsonStr) {

        try {
            IndexRequest request = new IndexRequest(index)
                    .id(id)
                    .source(jsonStr, XContentType.JSON);
            IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
            //System.out.println(indexResponse.toString());
            return indexResponse.getId();
        } catch (IOException e) {
            log.warn("同步保存doc失败, _index=[{}], _id=[{}], _body=[{}]", index, id, jsonStr);
            e.printStackTrace();
        }
        return index;
    }

    /**
     * 异步创建doc
     * @param index
     * @param id
     * @param jsonStr
     * @return
     */
    public void docSaveByJsonAsync(String index,String id,String jsonStr){
        IndexRequest request = new IndexRequest(index);
        request.id(id);
        request.source(jsonStr,XContentType.JSON);
        client.indexAsync(request, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                log.info("保存doc, _index=[{}], _id=[{}]成功, _version=[{}], _result=[{}]", index, indexResponse.getId(), indexResponse.getVersion(), indexResponse.getResult());
            }
            @Override
            public void onFailure(Exception e) {
                e.printStackTrace();
                log.warn("异步保存失败，尝试同步方式保存doc, ex=[{}]", e.getMessage());
                try {
                    IndexResponse response = client.index(request, RequestOptions.DEFAULT);
                    DocWriteResponse.Result result = response.getResult();
                    if(!(result==DocWriteResponse.Result.UPDATED || result==DocWriteResponse.Result.CREATED )){
                        log.warn("同步保存doc失败,_index=[{}], _id=[{}], _body=[{}]", index, id, jsonStr);
                    }
                }catch (IOException io){
                    io.printStackTrace();
                }
            }
        });
    }

    public String docSaveByMap(String index, String id, Map<String, Object> map) {
        try {
            IndexRequest request = new IndexRequest(index).id(id)
                    .source(map);
            IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
            return indexResponse.getId();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return index;
    }

    /**
     * 插入数据
     * @param index
     * @param id
     * @param object
     * @return
     */
    public String docSaveByJSIONObject(String index, String id, JSONObject object){
        IndexRequest request = new IndexRequest(index);
        request.id(id);
        try {
            request.source(JSON.toJSONString(object), XContentType.JSON);
            IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
            return  indexResponse.getId();
        }catch (Exception e){
            e.printStackTrace();
        }
        return  null;
    }

    /**
     * 根据id删除doc
     * @param index
     * @param id
     * @return
     */
    public Boolean docDeleteById(String index, String id) {
        try {
            DeleteRequest deleteRequest = new DeleteRequest(index, id);
            DeleteResponse delete = client.delete(deleteRequest, RequestOptions.DEFAULT);
            if(delete.status()== RestStatus.OK){
                log.info("DELETE /{}/_doc/{}/\r\n", index, id);
                return true;
            }
        }catch (IOException e){
            e.printStackTrace();
        }
        return false;
    }

    public void docDeleteByIdAsync(String index,String id){
        DeleteRequest request = new DeleteRequest(index, id);
        try {
            client.deleteAsync(request, RequestOptions.DEFAULT, new ActionListener<DeleteResponse>() {
                @Override
                public void onResponse(DeleteResponse deleteResponse) {
                    log.info("删除doc成功, _index=[{}], _id=[{}]", index, deleteResponse.getId());
                }

                @Override
                public void onFailure(Exception e) {
                    e.printStackTrace();
                    log.warn("删除doc失败, _index=[{}], _id=[{}]", index, id);
                }
            });
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public void docDeleteBulk(String index, List<ESBaseData> list){
        BulkRequest request = new BulkRequest();
        request.timeout(TimeValue.timeValueMinutes(10));
        for (int i = 0; i < list.size(); i++) {
            ESBaseData b = list.get(i);
            request.add(new DeleteRequest(index, b.getId()));
        }
        try {
            BulkResponse bulk = client.bulk(request, RequestOptions.DEFAULT);
            BulkItemResponse[] bulkItemResponses = bulk.getItems();
            int length = bulkItemResponses.length;
            for(int i = 0; i < length; ++i) {
                BulkItemResponse response = bulkItemResponses[i];
                if (response.isFailed()) {//查看所有请求失败结果
                    log.info("批量删除[{}]过程中,id为[{}]的删除失败,失败原因[{}]",response.getIndex(),response.getId(),response.getFailureMessage());
                }else{//请求成功的
                    log.info("批量删除[{}]过程中,id为[{}]的删除成功,状态[{}],version[{}]",response.getIndex(),response.getId(),response.status(),response.getVersion());
            }
            }
        }catch (IOException e){
            e.printStackTrace();
            log.warn("批量[{}]删除失败",index);
        }
    }

    public void docDeleteBulkAsync(String index,List<ESBaseData> list){
        BulkRequest request = new BulkRequest();
        request.timeout(TimeValue.timeValueMinutes(10));
        for (int i = 0; i < list.size(); i++) {
            ESBaseData b = list.get(i);
            request.add(new DeleteRequest(index, b.getId()));
        }
        client.bulkAsync(request, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse bulkItemResponses) {
                BulkItemResponse[] bulkItems = bulkItemResponses.getItems();
                int length = bulkItems.length;
                for(int i = 0; i < length; ++i) {
                    BulkItemResponse response = bulkItems[i];
                    if (response.isFailed()) {//查看所有请求失败结果
                        log.info("删除保存[{}]过程中,id为[{}]的删除失败,失败原因[{}]",response.getIndex(),response.getId(),response.getFailureMessage());
                    }else{//请求成功的
                        log.info("删除保存[{}]过程中,id为[{}]的删除成功,状态[{}],version[{}]",response.getIndex(),response.getId(),response.status(),response.getVersion());
                    }
                }
            }
            @Override
            public void onFailure(Exception e) {
                log.warn("批量[{}]删除失败,失败原因[{}]",index,e.getMessage());
            }
        });
    }

    /**
     * 根据条件删除
     * @param index
     * @param query
     * @return
     */
    public Long docDeleteByQuery(final String index, QueryBuilder query) {
        try {
            DeleteByQueryRequest request = new DeleteByQueryRequest(index);
            request.setQuery(query).setRefresh(true);
            BulkByScrollResponse bulkByScrollResponse = client.deleteByQuery(request, RequestOptions.DEFAULT);
            return bulkByScrollResponse.getDeleted();
        }catch (IOException e){
            e.printStackTrace();
            return null;
        }
    }

    public boolean docUpdateById(String index,String id ,String jsonStr){
        UpdateRequest request = new UpdateRequest(index, id);
        request.doc(jsonStr,XContentType.JSON);
        try {
            UpdateResponse response = client.update(request,RequestOptions.DEFAULT);
            return response.status() == RestStatus.OK;
        }catch (IOException e){
            e.printStackTrace();
            log.warn("更新doc失败, _index=[{}], _id=[{}],_jsonStr=[{}]", index, id,jsonStr);
        }
        return false;
    }


    /**
     * 根据index,id索引文件
     * @param index
     * @param id
     * @return
     */
    public Map<String, Object> searchDoc(String index,String id) {
        try {
            GetRequest searchRequest = new GetRequest(index,id);
            GetResponse documentFields = client.get(searchRequest, RequestOptions.DEFAULT);
            Map<String, Object> sourceAsMap = documentFields.getSourceAsMap();
            //sourceAsMap.put("score",documentFields.getSource());
            return  sourceAsMap;
        }catch (IOException e){
            log.warn("查询doc异常，index=[{}],id=[{}], ex=[{}]",index,id, e.getMessage());
        }
        return  null;
    }

    /**
     * 按条件查询数据
     * @param index
     * @param start
     * @param size
     * @param queryBuilder
     * @return
     */
    public Map<String, Object> searchByQuery(String index,int start, int size, QueryBuilder queryBuilder) {
        try {
            Map<String, Object> resultMap = new HashMap<>();
            SearchRequest searchRequest = new SearchRequest(index);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            if(queryBuilder !=null){
                searchSourceBuilder.query(queryBuilder);
            }
            if(start >=0 && size >=0){
                searchSourceBuilder.from(start);
                searchSourceBuilder.size(size);
            }
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = searchResponse.getHits();
            Long total = hits.getTotalHits().value;
            resultMap.put("total",total);
            SearchHit[] searchHits = hits.getHits();
            List<Map<String, Object>> mapList = new ArrayList<>(searchHits.length);
            for(SearchHit hit : searchHits){
                //存储的字段
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                //得分
                sourceAsMap.put("score",hit.getScore());
                mapList.add(sourceAsMap);
            }
            resultMap.put("data",mapList);
            return resultMap;
        }catch (IOException e){
            e.printStackTrace();
            return null;
        }
    }




    /**
     * 按条件查询数据
     * @param index
     * @param pageNo
     * @param pageSize
     * @param boolQueryBuilder
     * @return
     */
    public  Page<Map<String, Object>> searchPageByQuery(String index, int pageNo, int pageSize, BoolQueryBuilder  boolQueryBuilder) {
        try {

            Map<String, Object> resultMap = new HashMap<>();
            SearchRequest searchRequest = new SearchRequest(index);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.trackTotalHits(true);
            if(boolQueryBuilder !=null){
                searchSourceBuilder.query(boolQueryBuilder);
            }
            if(pageNo >=0 && pageSize >=0){
                searchSourceBuilder.from((pageNo-1)*pageSize);
                searchSourceBuilder.size(pageSize);
            }
            searchRequest.source(searchSourceBuilder);

            long t1 = System.currentTimeMillis();
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            log.debug("处理查询："+boolQueryBuilder.toString()+";\n耗时："+((System.currentTimeMillis()-t1)));
            SearchHits hits = searchResponse.getHits();
            Long total = hits.getTotalHits().value;
            SearchHit[] searchHits = hits.getHits();
            List<Map<String, Object>> mapList = new ArrayList<>(searchHits.length);
            for(SearchHit hit : searchHits){
                //存储的字段
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                sourceAsMap.put("id",hit.getId());
                //得分
                sourceAsMap.put("score",hit.getScore());
                mapList.add(sourceAsMap);
            }
            resultMap.put("data",mapList);
            Page<Map<String, Object>> page = new Page<Map<String, Object>>((pageNo-1)*pageSize+1, pageSize);
            page.setTotal(total);
            page.setRecords(mapList);
            return   page ;
        }catch (IOException e){
            e.printStackTrace();
            return null;
        }
    }

}
