package com.zjty.dev.device.es.subject.service.impl;

import com.zjty.dev.device.es.subject.entity.CollectionRs;
import com.zjty.dev.device.es.subject.entity.EsSource;
import com.zjty.dev.device.es.subject.service.EsUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author LJJ cnljj1995@gmail.com
 * on 1/23/21
 */
@Service
@Slf4j
public class EsUtilImpl implements EsUtil {

    @Override
    public boolean createIndex(String indexName, Map<String, Object> mapping) {

        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("localhost", 9200, "http")));


        CreateIndexRequest request = new CreateIndexRequest(indexName);
        request.settings(Settings.builder()
                .put("index.number_of_shards", 3)
                .put("index.number_of_replicas", 2)
                .build());

        request.mapping("doc", mapping);

        try {
            client.indices().create(request, RequestOptions.DEFAULT);
            client.close();
            log.info("[es] 创建索引成功,被创建的索引为:{}", indexName);
            return true;
        } catch (Exception e) {
            log.error("[es] 创建索引出错，错误信息:{}", e.getMessage());
            try {
                client.close();
            } catch (IOException ex) {
                log.error("[es] es client关闭异常，错误信息:{}", ex.getMessage());
            }
            return false;
        }
    }

    @Override
    public void bulkAdd(String indexName, List<EsSource> data) {

        // es的bulk方法不适合无限制的保存，此处预设为每次保存50条，具体需根据服务器性能进行测试
        int limit = 50;
        int size = data.size();
        int page = size / limit;
        boolean tag = size > page * limit;

        for (int i = 0; i < page; i++) {
            addByBulk(indexName, data.subList(limit * i, limit * (i + 1)));
            log.info("[es] 批量保存进度:{}/{}", i + 1, tag ? page + 1 : page);
        }

        if (size > page * limit) {
            addByBulk(indexName, data.subList(page * limit, size));
            log.info("[es] 批量保存进度:{}/{}", page + 1, page + 1);
        }

    }

    @Override
    public void bulkDel(String indexName, List<String> ids) {

        // es的bulk方法不适合无限制的保存，此处预设为每次保存50条，具体需根据服务器性能进行测试
        int limit = 50;
        int size = ids.size();
        int page = size / limit;
        boolean tag = size > page * limit;

        for (int i = 0; i < page; i++) {
            delByBulk(indexName, ids.subList(limit * i, limit * (i + 1)));
            log.info("[es] 批量删除进度:{}/{}", i + 1, tag ? page + 1 : page);
        }

        if (size > page * limit) {
            delByBulk(indexName, ids.subList(page * limit, size));
            log.info("[es] 批量删除进度:{}/{}", page + 1, page + 1);
        }
    }

    @Override
    public CollectionRs search(String indexName, String key, int page, int size) {

        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("localhost", 9200, "http")));

        SearchRequest searchRequest = new SearchRequest(indexName);

        SearchSourceBuilder builder = new SearchSourceBuilder().query(
                QueryBuilders.queryStringQuery(key));
        builder.size(size)
                .from(size * page)
                .sort("updateTime", SortOrder.DESC);
        searchRequest.source(builder);

        SearchResponse search = null;
        try {
            search = client.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("[es] 关键字搜索出错:{}", e.getMessage());
        }
        try {
            client.close();
        } catch (IOException e) {
            log.error("[es] 关闭客户端出错:{}", e.getMessage());
        }

        List<Map<String, Object>> rs = new ArrayList<>();
        List<String> ids = new ArrayList<>();
        for (SearchHit hit : search.getHits().getHits()) {
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            sourceAsMap.put("_id",hit.getId());
            rs.add(sourceAsMap);
            ids.add(hit.getId());
        }
        CollectionRs collectionRs = new CollectionRs(search.getHits().getTotalHits().value, ids, rs);
        return collectionRs;

    }

    private void delByBulk(String indexName, List<String> ids) {
        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("localhost", 9200, "http")));

        BulkRequest bulkRequest = new BulkRequest();
        ids.forEach(o -> bulkRequest.add(new DeleteRequest(indexName, o)));
        try {
            client.bulk(bulkRequest, RequestOptions.DEFAULT);
            client.close();
        } catch (Exception e) {
            log.error("[es] 批量删除数据出错: {},出错信息: {}", indexName, e.getMessage());
            try {
                client.close();
            } catch (IOException ex) {
                log.error("[es] es client关闭异常，错误信息:{}", ex.getMessage());

            }
        }
    }

    private void addByBulk(String indexName, List<EsSource> data) {
        RestHighLevelClient client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("localhost", 9200, "http")));

        BulkRequest bulkRequest = new BulkRequest();
        data.forEach(o -> bulkRequest.add(new IndexRequest(indexName, "doc", o.getId()).source(o.getJsonMap())));

        try {
            client.bulk(bulkRequest, RequestOptions.DEFAULT);
            client.close();
        } catch (Exception e) {
            log.error("[es] 批量插入数据出错: {},出错信息: {}", indexName, e.getMessage());
            try {
                client.close();
            } catch (IOException ex) {
                log.error("[es] es client关闭异常，错误信息:{}", ex.getMessage());

            }
        }
    }

}
